feat(auth): harden credential storage and secure tui bearer input

pyproject.toml

@@ -45,6 +45,7 @@ dependencies = [
     "textual-serve>=1.1.0",
     "mcp[cli]>=1.2.0",
     "truststore>=0.9.0",
+    "keyring>=25.6.0",
 ]
 
 [dependency-groups]

src/a2a_handler/cli/__init__.py

@@ -14,13 +14,16 @@
 truststore.inject_into_ssl()
 
 import logging
+import shlex
+import subprocess
 
 logging.getLogger().setLevel(logging.WARNING)
 
 import rich_click as click
 
 from a2a_handler import __version__
 from a2a_handler.common import Output, configure_output, get_logger, setup_logging
+from a2a_handler.common.input_validation import reject_control_chars
 from a2a_handler.common.output import OutputFormat
 from a2a_handler.tui import HandlerTUI
 
@@ -99,12 +102,74 @@ def version(ctx: click.Context) -> None:
 
 
 @cli.command()
-@click.option("--bearer", "-b", "bearer_token", help="Bearer token for agent auth")
-def tui(bearer_token: str | None) -> None:
+@click.option(
+    "--bearer",
+    "-b",
+    "bearer_token",
+    help="Temporary bearer token for this TUI run (not saved)",
+)
+@click.option(
+    "--bearer-command",
+    help="Command that prints a temporary bearer token to stdout",
+)
+@click.option(
+    "--bearer-stdin",
+    is_flag=True,
+    help="Read a temporary bearer token from stdin",
+)
+def tui(
+    bearer_token: str | None,
+    bearer_command: str | None,
+    bearer_stdin: bool,
+) -> None:
     """Launch the interactive terminal interface."""
+    auth_sources = [
+        bool(bearer_token),
+        bool(bearer_command),
+        bearer_stdin,
+    ]
+    if sum(auth_sources) > 1:
+        raise click.ClickException(
+            "Use only one of --bearer, --bearer-command, or --bearer-stdin"
+        )
+
+    resolved_bearer_token = bearer_token
+    if bearer_command:
+        reject_control_chars(bearer_command, "bearer_command")
+        try:
+            command_parts = shlex.split(bearer_command)
+            if not command_parts:
+                raise click.ClickException("--bearer-command cannot be empty")
+            command_result = subprocess.run(
+                command_parts,
+                check=True,
+                capture_output=True,
+                text=True,
+            )
+        except ValueError as error:
+            raise click.ClickException(f"Invalid --bearer-command: {error}") from error
+        except subprocess.CalledProcessError as error:
+            stderr = error.stderr.strip() if error.stderr else ""
+            message = f"--bearer-command failed with exit code {error.returncode}"
+            if stderr:
+                message = f"{message}: {stderr}"
+            raise click.ClickException(message) from error
+
+        resolved_bearer_token = command_result.stdout.strip()
+        if not resolved_bearer_token:
+            raise click.ClickException("--bearer-command produced an empty token")
+
+    elif bearer_stdin:
+        resolved_bearer_token = click.get_text_stream("stdin").read().strip()
+        if not resolved_bearer_token:
+            raise click.ClickException("--bearer-stdin received an empty token")
+
+    if resolved_bearer_token:
+        reject_control_chars(resolved_bearer_token, "bearer_token")
+
     log.info("Launching TUI")
     logging.getLogger().handlers = []
-    app = HandlerTUI(initial_bearer_token=bearer_token)
+    app = HandlerTUI(initial_bearer_token=resolved_bearer_token)
     app.run()
 
 

src/a2a_handler/cli/_config.py

@@ -67,6 +67,12 @@
             "options": ["--bearer", "--api-key", "--api-key-header"],
         },
     ],
+    "handler auth source set": [
+        {
+            "name": "Source Options",
+            "options": ["--provider", "--command"],
+        },
+    ],
 }
 
 click.rich_click.COMMAND_GROUPS = {
@@ -97,6 +103,12 @@
         {"name": "Session Commands", "commands": ["list", "show", "clear"]},
     ],
     "handler auth": [
-        {"name": "Auth Commands", "commands": ["set", "show", "clear"]},
+        {
+            "name": "Auth Commands",
+            "commands": ["set", "show", "clear", "source"],
+        },
+    ],
+    "handler auth source": [
+        {"name": "Source Commands", "commands": ["set", "show", "clear"]},
     ],
 }

src/a2a_handler/cli/auth.py

@@ -6,11 +6,22 @@
 
 from a2a_handler.auth import AuthType, create_api_key_auth, create_bearer_auth
 from a2a_handler.common import Output
+from a2a_handler.common import (
+    clear_agent_bearer_command,
+    get_agent_bearer_command,
+    get_default_bearer_command,
+    save_agent_bearer_command,
+    save_default_bearer_command,
+)
 from a2a_handler.common.input_validation import (
     InputValidationError,
     reject_control_chars,
     validate_agent_url,
 )
+from a2a_handler.credentials import (
+    BUILTIN_BEARER_PROVIDERS,
+    get_builtin_provider_command,
+)
 from a2a_handler.session import clear_credentials, get_credentials, set_credentials
 
 from ._helpers import handle_validation_error
@@ -70,7 +81,7 @@ def auth_set(
 
     set_credentials(agent_url, credentials)
 
-    output.success(f"Set {auth_type_display} for {agent_url}")
+    output.success(f"Set {auth_type_display} for {agent_url} (saved to OS keyring)")
 
 
 @auth.command("show")
@@ -117,3 +128,130 @@ def auth_clear(agent_url: str) -> None:
 
     clear_credentials(agent_url)
     output.success(f"Cleared credentials for {agent_url}")
+
+
+@auth.group("source")
+def auth_source() -> None:
+    """Manage automatic bearer token sources."""
+    pass
+
+
+@auth_source.command("set")
+@click.argument("agent_url", required=False)
+@click.option(
+    "--provider",
+    type=click.Choice(sorted(BUILTIN_BEARER_PROVIDERS.keys())),
+    help="Built-in provider for bearer tokens (for example: gcloud)",
+)
+@click.option(
+    "--command",
+    "bearer_command",
+    help="Command that prints a bearer token to stdout",
+)
+def auth_source_set(
+    agent_url: str | None,
+    provider: str | None,
+    bearer_command: str | None,
+) -> None:
+    """Set an automatic bearer token command for an agent or globally.
+
+    If AGENT_URL is omitted, sets the global default source.
+    """
+    output = Output()
+    try:
+        if agent_url:
+            validate_agent_url(agent_url)
+        if provider and bearer_command:
+            raise InputValidationError(
+                code="invalid_auth_source_arguments",
+                message="Provide either --provider or --command, not both",
+                suggestion="Pick one auth source mode",
+            )
+        if not provider and not bearer_command:
+            raise InputValidationError(
+                code="missing_auth_source_arguments",
+                message="Provide --provider or --command",
+                suggestion="For gcloud use: --provider gcloud",
+            )
+        if bearer_command:
+            reject_control_chars(bearer_command, "bearer_command")
+    except InputValidationError as error:
+        handle_validation_error(error, output)
+        raise click.Abort() from error
+
+    command = bearer_command
+    if provider:
+        command = get_builtin_provider_command(provider)
+        if command is None:
+            output.error(f"Unsupported provider: {provider}")
+            raise click.Abort()
+
+    assert command is not None
+    if agent_url:
+        save_agent_bearer_command(agent_url, command)
+        output.success(f"Set auth source for {agent_url}")
+    else:
+        save_default_bearer_command(command)
+        output.success("Set global auth source")
+
+    if provider:
+        output.field("Provider", provider)
+    output.field("Command", command)
+
+
+@auth_source.command("show")
+@click.argument("agent_url", required=False)
+def auth_source_show(agent_url: str | None) -> None:
+    """Show configured automatic bearer token sources."""
+    output = Output()
+    try:
+        if agent_url:
+            validate_agent_url(agent_url)
+    except InputValidationError as error:
+        handle_validation_error(error, output)
+        raise click.Abort() from error
+
+    output.header("Auth Source")
+
+    if agent_url:
+        command = get_agent_bearer_command(agent_url)
+        if command:
+            output.field("Scope", f"Agent: {agent_url}")
+            output.field("Command", command)
+            return
+
+        default_command = get_default_bearer_command()
+        if default_command:
+            output.field("Scope", f"Default fallback for {agent_url}")
+            output.field("Command", default_command)
+            return
+
+        output.dim("No auth source configured")
+        return
+
+    command = get_default_bearer_command()
+    if command:
+        output.field("Scope", "Global")
+        output.field("Command", command)
+    else:
+        output.dim("No global auth source configured")
+
+
+@auth_source.command("clear")
+@click.argument("agent_url", required=False)
+def auth_source_clear(agent_url: str | None) -> None:
+    """Clear automatic bearer token source for an agent or globally."""
+    output = Output()
+    try:
+        if agent_url:
+            validate_agent_url(agent_url)
+    except InputValidationError as error:
+        handle_validation_error(error, output)
+        raise click.Abort() from error
+
+    if agent_url:
+        clear_agent_bearer_command(agent_url)
+        output.success(f"Cleared auth source for {agent_url}")
+    else:
+        save_default_bearer_command(None)
+        output.success("Cleared global auth source")

src/a2a_handler/cli/card.py

@@ -8,8 +8,8 @@
 
 from a2a_handler.common import Output, get_logger
 from a2a_handler.common.input_validation import InputValidationError, validate_agent_url
+from a2a_handler.credentials import resolve_auth_credentials
 from a2a_handler.service import A2AService
-from a2a_handler.session import get_credentials
 from a2a_handler.validation import (
     ValidationResult,
     validate_agent_card_from_file,
@@ -46,9 +46,15 @@ def card_get(agent_url: str, authenticated: bool) -> None:
         raise click.Abort() from error
 
     log.info("Fetching agent card from %s", agent_url)
-    credentials = get_credentials(agent_url) if authenticated else None
-    if authenticated and credentials is None:
-        log.warning("No saved credentials found for %s", agent_url)
+    credentials = None
+    if authenticated:
+        try:
+            credentials = resolve_auth_credentials(agent_url)
+        except InputValidationError as error:
+            handle_validation_error(error, output)
+            raise click.Abort() from error
+        if credentials is None:
+            log.warning("No credentials or auth source found for %s", agent_url)
 
     async def do_get() -> None:
         try:

src/a2a_handler/cli/message.py

@@ -6,7 +6,7 @@
 
 import rich_click as click
 
-from a2a_handler.auth import AuthCredentials, create_api_key_auth, create_bearer_auth
+from a2a_handler.auth import AuthCredentials
 from a2a_handler.common import Output, get_logger
 from a2a_handler.common.input_validation import (
     InputValidationError,
@@ -17,8 +17,9 @@
     validate_resource_id,
     validate_webhook_url,
 )
+from a2a_handler.credentials import resolve_auth_credentials
 from a2a_handler.service import A2AService, SendResult
-from a2a_handler.session import get_credentials, get_session, update_session
+from a2a_handler.session import get_session, update_session
 
 from ._helpers import build_http_client, handle_client_error, handle_validation_error
 
@@ -154,12 +155,15 @@ def message_send(
             log.info("Using saved context: %s", context_id)
 
     credentials: AuthCredentials | None = None
-    if bearer_token:
-        credentials = create_bearer_auth(bearer_token)
-    elif api_key:
-        credentials = create_api_key_auth(api_key)
-    else:
-        credentials = get_credentials(agent_url)
+    try:
+        credentials = resolve_auth_credentials(
+            agent_url,
+            bearer_token=bearer_token,
+            api_key=api_key,
+        )
+    except InputValidationError as error:
+        handle_validation_error(error, output)
+        raise click.Abort() from error
 
     async def do_send() -> None:
         try:
@@ -272,6 +276,9 @@ async def _stream_message(
         output.line(
             "Set credentials with: handler auth set <agent_url> --bearer <token>"
         )
+        output.line(
+            "Or configure automatic tokens: handler auth source set --provider gcloud"
+        )
 
 
 def _format_send_result(result: SendResult, output: Output) -> None:
@@ -294,6 +301,9 @@ def _format_send_result(result: SendResult, output: Output) -> None:
         output.line(
             "Or provide inline: handler message send <agent_url> --bearer <token> ..."
         )
+        output.line(
+            "Or configure automatic tokens: handler auth source set --provider gcloud"
+        )
     elif result.text:
         output.markdown(result.text)
     else: