Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 39 additions & 6 deletions src/crux_cli/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,12 @@
from crux_cli.secrets import get_backend, load_secrets_index


def auth_single(mcp_name: str, registry: dict[str, Any] | None = None) -> None:
def auth_single(
mcp_name: str,
registry: dict[str, Any] | None = None,
*,
inline_values: dict[str, str] | None = None,
) -> None:
"""Authenticate a single MCP by dispatching to the appropriate auth handler."""
if registry is None:
registry = load_registry()
Expand All @@ -40,13 +45,13 @@ def auth_single(mcp_name: str, registry: dict[str, Any] | None = None) -> None:
auth_type = "setup-cmd"

if auth_type == "keychain":
_auth_keychain(mcp_name, auth)
_auth_keychain(mcp_name, auth, inline_values=inline_values)
elif auth_type == "external-cli":
_auth_external_cli(mcp_name, auth)
elif auth_type == "setup-cmd":
_auth_setup_cmd(mcp_name, auth)
elif auth_type == "bearer":
_auth_bearer(mcp_name, auth)
_auth_bearer(mcp_name, auth, inline_values=inline_values)
elif auth_type in ("oauth", "oauth-client-credentials"):
# Defer to oauth module (Plan phase 3)
from crux_cli.oauth import run_client_credentials_flow, run_oauth_flow
Expand Down Expand Up @@ -106,8 +111,13 @@ def auth_all(registry: dict[str, Any] | None = None) -> None:
print(f"\u26a0\ufe0f Failed: {e}")


def _auth_keychain(mcp_name: str, auth: dict[str, Any]) -> None:
"""Authenticate via OS keychain — prompt for each env var."""
def _auth_keychain(
mcp_name: str,
auth: dict[str, Any],
*,
inline_values: dict[str, str] | None = None,
) -> None:
"""Authenticate via OS keychain — prompt for each env var (or use inline values)."""
env_vars = auth.get("env_vars", [])
if not env_vars:
print(f"\u2139\ufe0f No env vars configured for '{mcp_name}'")
Expand All @@ -118,12 +128,24 @@ def _auth_keychain(mcp_name: str, auth: dict[str, Any]) -> None:
stored = secrets_index.get(mcp_name, [])

for var in env_vars:
# Use inline value if provided (always overwrite)
if inline_values and var in inline_values:
backend.set(mcp_name, var, inline_values[var])
print(f" \u2705 Stored {var}")
continue

# Skip already-stored secrets in interactive mode
if var in stored:
existing = backend.get(mcp_name, var)
if existing:
print(f" {var}: already set (use crux mcp auth {mcp_name} to reset)")
continue

# If inline_values were provided but this var is missing, skip prompting
if inline_values is not None:
print(f" Skipped {var} (no --value provided)")
continue

value = getpass.getpass(f" Enter {var}: ")
if not value:
print(f" Skipped {var}")
Expand Down Expand Up @@ -190,11 +212,22 @@ def _auth_setup_cmd(mcp_name: str, auth: dict[str, Any]) -> None:
sys.exit(result.returncode)


def _auth_bearer(mcp_name: str, auth: dict[str, Any]) -> None:
def _auth_bearer(
mcp_name: str,
auth: dict[str, Any],
*,
inline_values: dict[str, str] | None = None,
) -> None:
"""Authenticate via static Bearer token."""
keychain_key = auth.get("keychain_key", "API_TOKEN")
backend = get_backend()

# Use inline value if provided (always overwrite)
if inline_values and keychain_key in inline_values:
backend.set(mcp_name, keychain_key, inline_values[keychain_key])
print(f"\u2705 Token stored for '{mcp_name}'")
return

existing = backend.get(mcp_name, keychain_key)
if existing:
replace = input(f" Token already stored for '{mcp_name}'. Replace? [y/N] ").strip().lower()
Expand Down
14 changes: 13 additions & 1 deletion src/crux_cli/cli/commands/mcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -481,8 +481,20 @@ def cmd_mcp_auth(args: argparse.Namespace) -> None:
name = getattr(args, "name", None)
do_all = getattr(args, "all", False)

values = getattr(args, "value", None)

if name:
auth_single(name)
# Parse --value KEY=VALUE pairs into a dict
inline_values: dict[str, str] | None = None
if values:
inline_values = {}
for pair in values:
if "=" not in pair:
print(f"\u274c Invalid --value format: '{pair}' (expected KEY=VALUE)")
sys.exit(1)
k, v = pair.split("=", 1)
inline_values[k] = v
auth_single(name, inline_values=inline_values)
elif do_all:
auth_all()
else:
Expand Down
1 change: 1 addition & 0 deletions src/crux_cli/cli/main.py
Original file line number Diff line number Diff line change
Expand Up @@ -82,6 +82,7 @@ def main() -> None:
p.add_argument("name", nargs="?", help="MCP name (omit to show auth status)")
p.add_argument("--all", action="store_true", help="Authenticate all MCPs needing auth")
p.add_argument("--refresh", action="store_true", help="Refresh OAuth token only")
p.add_argument("--value", action="append", metavar="KEY=VALUE", help="Set secret non-interactively (repeatable)")
p.set_defaults(func=cmd_mcp_auth)

p = mcp_sub.add_parser("status", help="Probe all registered MCP servers")
Expand Down
20 changes: 20 additions & 0 deletions tests/integration/test_cli_mcp.py
Original file line number Diff line number Diff line change
Expand Up @@ -69,6 +69,26 @@ def test_remove_nonexistent_fails(self, crux_env):
assert result.returncode != 0


@pytest.mark.integration
class TestMcpAuth:
def test_auth_value_flag_accepted(self, crux_env):
"""--value flag is accepted by the CLI parser."""
env, root = crux_env
run_crux("mcp", "add", "authed", "--npx", "@test/authed", "--keychain", "API_KEY", env=env)
# --value should be accepted (may fail to store due to keychain, but shouldn't be a parse error)
result = run_crux("mcp", "auth", "authed", "--value", "API_KEY=test123", env=env)
# Should not fail with argparse error (exit code 2)
assert result.returncode != 2

def test_auth_value_invalid_format_fails(self, crux_env):
"""--value without = should fail."""
env, root = crux_env
run_crux("mcp", "add", "authed2", "--npx", "@test/authed2", "--keychain", "KEY", env=env)
result = run_crux("mcp", "auth", "authed2", "--value", "NOEQUALS", env=env)
assert result.returncode != 0
assert "Invalid --value format" in result.stdout


@pytest.mark.integration
class TestMcpList:
def test_list_shows_mcps(self, crux_env):
Expand Down
79 changes: 79 additions & 0 deletions tests/unit/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -250,6 +250,85 @@ def test_auth_single_mcp_not_found(self, monkeypatch, capsys):
assert "nonexistent" in captured.out


# ---------------------------------------------------------------------------
# --value inline auth tests
# ---------------------------------------------------------------------------


class TestInlineValueAuth:
"""Test that --value flag sets secrets non-interactively."""

def test_keychain_inline_values_stored(self, monkeypatch):
"""Inline values are stored without prompting."""
monkeypatch.setattr("crux_cli.auth.load_secrets_index", lambda: {})
mock_backend = MockBackend()
monkeypatch.setattr("crux_cli.auth.get_backend", lambda: mock_backend)

from crux_cli.auth import auth_single

registry = _make_registry("mymcp", {"type": "keychain", "env_vars": ["API_KEY", "SECRET"]})
auth_single("mymcp", registry, inline_values={"API_KEY": "key123", "SECRET": "sec456"})

assert len(mock_backend.set_calls) == 2
stored = {call[1]: call[2] for call in mock_backend.set_calls}
assert stored["API_KEY"] == "key123"
assert stored["SECRET"] == "sec456" # noqa: S105

def test_keychain_inline_overwrites_existing(self, monkeypatch):
"""Inline values overwrite already-stored secrets."""
monkeypatch.setattr("crux_cli.auth.load_secrets_index", lambda: {"mymcp": ["API_KEY"]})
mock_backend = MockBackend()
mock_backend._data[("mymcp", "API_KEY")] = "old-value"
monkeypatch.setattr("crux_cli.auth.get_backend", lambda: mock_backend)

from crux_cli.auth import _auth_keychain

_auth_keychain("mymcp", {"env_vars": ["API_KEY"]}, inline_values={"API_KEY": "new-value"})

assert len(mock_backend.set_calls) == 1
assert mock_backend.set_calls[0][2] == "new-value"

def test_keychain_inline_partial_skips_missing(self, monkeypatch, capsys):
"""When inline_values provided but a var is missing, it's skipped without prompting."""
monkeypatch.setattr("crux_cli.auth.load_secrets_index", lambda: {})
mock_backend = MockBackend()
monkeypatch.setattr("crux_cli.auth.get_backend", lambda: mock_backend)

from crux_cli.auth import _auth_keychain

_auth_keychain("mymcp", {"env_vars": ["API_KEY", "SECRET"]}, inline_values={"API_KEY": "key123"})

assert len(mock_backend.set_calls) == 1
assert mock_backend.set_calls[0][1] == "API_KEY"
captured = capsys.readouterr()
assert "Skipped SECRET" in captured.out

def test_bearer_inline_value_stored(self, monkeypatch):
"""Inline value for bearer token is stored without prompting."""
mock_backend = MockBackend()
monkeypatch.setattr("crux_cli.auth.get_backend", lambda: mock_backend)

from crux_cli.auth import _auth_bearer

_auth_bearer("mymcp", {"keychain_key": "API_TOKEN"}, inline_values={"API_TOKEN": "mytoken"})

assert len(mock_backend.set_calls) == 1
assert mock_backend.set_calls[0][2] == "mytoken"

def test_bearer_inline_overwrites_existing(self, monkeypatch):
"""Inline bearer token overwrites existing without prompting."""
mock_backend = MockBackend()
mock_backend._data[("mymcp", "API_TOKEN")] = "old-token"
monkeypatch.setattr("crux_cli.auth.get_backend", lambda: mock_backend)

from crux_cli.auth import _auth_bearer

_auth_bearer("mymcp", {"keychain_key": "API_TOKEN"}, inline_values={"API_TOKEN": "new-token"})

assert len(mock_backend.set_calls) == 1
assert mock_backend.set_calls[0][2] == "new-token"


# ---------------------------------------------------------------------------
# Inline auth during mcp add
# ---------------------------------------------------------------------------
Expand Down
Loading