refactor: simplify token resolution — remove legacy paths and JWT validation

- Remove legacy credential file paths (_creds_candidates)
- Remove JWT validation from _load_rnk_token (issuer/expiry without
  signature verification is security theater; the data API validates)
- Replace bare except Exception with specific types (bandit B110/B112)
- Add tests for _load_rnk_token and _resolve_token

Author: rokroskar

bases/renku_data_services/mcp_api/main.py

@@ -13,19 +13,16 @@
 Token resolution order
 ----------------------
 1. RENKU_ACCESS_TOKEN, RENKU_TOKEN, or RENKU_CLI_ACCESS_TOKEN env var.
-2. Legacy credential files (~/.config/renku-agent-skill/credentials.json, etc.).
-3. Official rnk CLI token file (platform-specific path, validated for issuer + expiry).
+2. rnk CLI token file (platform-specific path, token forwarded as-is to the data API).
 """
 
 from __future__ import annotations
 
 import asyncio
-import base64
 import json
 import logging
 import os
 import sys
-import time
 from collections.abc import Callable
 from pathlib import Path
 from typing import Any
@@ -45,18 +42,6 @@ def _base_url() -> str:
     return os.environ.get("RENKU_BASE_URL", "https://renkulab.io").rstrip("/")
 
 
-def _creds_candidates() -> list[Path]:
-    """Legacy credential file paths to search."""
-    candidates: list[Path] = []
-    if d := os.environ.get("RENKU_CONFIG_DIR"):
-        candidates.append(Path(d) / "credentials.json")
-    home = Path.home()
-    candidates += [
-        home / ".config" / "renku-agent-skill" / "credentials.json",
-        home / ".pi" / "renku-config" / "credentials.json",
-    ]
-    return candidates
-
 
 def _rnk_token_paths() -> list[Path]:
     """Paths where the official rnk CLI stores its token file (platform-specific)."""
@@ -72,62 +57,40 @@ def _rnk_token_paths() -> list[Path]:
 
 
 def _load_rnk_token() -> str | None:
-    """Read an access token from the rnk CLI token file, validating issuer and expiry."""
-    expected_issuer = _base_url() + "/auth/realms/Renku"
+    """Read an access token from the rnk CLI token file.
+
+    No JWT validation is performed here — the token is forwarded to the Renku
+    data API which is the authoritative validator (signature, issuer, expiry).
+    """
     for path in _rnk_token_paths():
         if not path.exists():
             continue
         try:
             data = json.loads(path.read_text())
             response = data.get("response") or data
             access = response.get("access_token")
-            if not access:
-                continue
-            try:
-                payload_part = access.split(".")[1]
-                payload_part += "=" * (-len(payload_part) % 4)
-                payload: dict[str, Any] = json.loads(base64.urlsafe_b64decode(payload_part.encode()))
-                if payload.get("iss") != expected_issuer:
-                    continue  # token is for a different deployment
-                if payload.get("exp") and time.time() > int(payload["exp"]) - 60:
-                    continue  # token is expired or expires in < 60 s
-            except (ValueError, KeyError, IndexError):
-                # JWT decode failed — accept the token anyway and let Keycloak validate it.
-                pass
-            return access
+            if access:
+                return access
         except (OSError, json.JSONDecodeError, KeyError):
             continue
     return None
 
 
 def _resolve_token() -> str:
     """Return the best available token, or raise with a helpful message."""
-    # 1. Environment variables (highest priority)
+    # 1. Environment variables (explicit override)
     for var in ("RENKU_ACCESS_TOKEN", "RENKU_TOKEN", "RENKU_CLI_ACCESS_TOKEN"):
         if t := os.environ.get(var):
             return t
 
-    # 2. Legacy credential files
-    checked: list[str] = []
-    for f in _creds_candidates():
-        checked.append(str(f))
-        if f.exists():
-            try:
-                entry = json.loads(f.read_text()).get(_base_url(), {})
-                if t := entry.get("access_token") or entry.get("token"):
-                    return t
-            except (OSError, json.JSONDecodeError, KeyError, AttributeError):
-                pass  # malformed or unreadable credential file — try next source
-
-    # 3. Official rnk CLI token file
+    # 2. rnk CLI token file
     if t := _load_rnk_token():
         return t
 
     rnk_paths = [str(p) for p in _rnk_token_paths()]
     raise RuntimeError(
         f"Not authenticated for {_base_url()}.\n"
         f"Run: rnk login\n"
-        f"Credentials searched in: {', '.join(checked)}\n"
         f"rnk token paths searched: {', '.join(rnk_paths)}\n"
         f"Or set RENKU_ACCESS_TOKEN in the MCP server environment config."
     )

pyproject.toml

@@ -295,7 +295,9 @@ disable_error_code = ["no-untyped-def", "var-annotated", "import-untyped"]
 # would require casts on every tool, adding noise without safety benefit.
 module = ["renku_data_services.mcp_api.*"]
 warn_return_any = false
-disable_error_code = ["no-untyped-def"]
+# misc covers "untyped decorator makes function untyped" from @mcp.tool() —
+# the mcp package doesn't ship type stubs.
+disable_error_code = ["no-untyped-def", "misc"]
 
 [[tool.mypy.overrides]]
 module = ["mcp.*", "fastmcp.*", "uvicorn.*", "starlette.*"]

test/bases/renku_data_services/mcp_api/test_mcp_server.py

@@ -2,12 +2,19 @@
 
 from __future__ import annotations
 
+import json
 from typing import Any
 
 import pytest
 
 from renku_data_services.mcp_api.dependencies import MCPDependencies
-from renku_data_services.mcp_api.main import _authorization_server_doc, _protected_resource_doc
+from renku_data_services.mcp_api.main import (
+    _authorization_server_doc,
+    _load_rnk_token,
+    _protected_resource_doc,
+    _resolve_token,
+    _rnk_token_paths,
+)
 from renku_data_services.mcp_api.server import (
     _admin_cache,
     _is_stale_session,
@@ -131,6 +138,86 @@ async def test_authorization_server_doc_no_keycloak_url():
     assert status == 503
 
 
+# ------------------------------------------------------------------ #
+# Token resolution — _load_rnk_token and _resolve_token               #
+# ------------------------------------------------------------------ #
+
+
+def test_load_rnk_token_finds_token_in_response_key(tmp_path, monkeypatch):
+    """_load_rnk_token reads the token from the 'response' wrapper used by rnk."""
+    token_file = tmp_path / "token.json"
+    token_file.write_text(json.dumps({"response": {"access_token": "my-token"}}))
+
+    monkeypatch.setattr("renku_data_services.mcp_api.main._rnk_token_paths", lambda: [token_file])
+    assert _load_rnk_token() == "my-token"
+
+
+def test_load_rnk_token_finds_token_at_root(tmp_path, monkeypatch):
+    """_load_rnk_token also reads the token when there's no 'response' wrapper."""
+    token_file = tmp_path / "token.json"
+    token_file.write_text(json.dumps({"access_token": "my-token"}))
+
+    monkeypatch.setattr("renku_data_services.mcp_api.main._rnk_token_paths", lambda: [token_file])
+    assert _load_rnk_token() == "my-token"
+
+
+def test_load_rnk_token_forwards_any_token_value(tmp_path, monkeypatch):
+    """_load_rnk_token forwards tokens as-is without JWT validation — the API validates."""
+    token_file = tmp_path / "token.json"
+    token_file.write_text(json.dumps({"access_token": "opaque-or-expired-or-wrong-issuer"}))
+
+    monkeypatch.setattr("renku_data_services.mcp_api.main._rnk_token_paths", lambda: [token_file])
+    assert _load_rnk_token() == "opaque-or-expired-or-wrong-issuer"
+
+
+def test_load_rnk_token_no_file(monkeypatch):
+    """_load_rnk_token returns None when no token file exists."""
+    monkeypatch.setattr("renku_data_services.mcp_api.main._rnk_token_paths", lambda: [])
+    assert _load_rnk_token() is None
+
+
+def test_resolve_token_prefers_env_var(tmp_path, monkeypatch):
+    """_resolve_token returns the env var even when an rnk token file exists."""
+    monkeypatch.setenv("RENKU_ACCESS_TOKEN", "env-token")
+    monkeypatch.setattr("renku_data_services.mcp_api.main._rnk_token_paths", lambda: [])
+    assert _resolve_token() == "env-token"
+
+
+def test_resolve_token_falls_back_to_rnk(tmp_path, monkeypatch):
+    """_resolve_token falls back to the rnk file when no env var is set."""
+    base_url = "https://renkulab.io"
+    monkeypatch.setenv("RENKU_BASE_URL", base_url)
+    monkeypatch.delenv("RENKU_ACCESS_TOKEN", raising=False)
+    monkeypatch.delenv("RENKU_TOKEN", raising=False)
+    monkeypatch.delenv("RENKU_CLI_ACCESS_TOKEN", raising=False)
+
+    token_file = tmp_path / "token.json"
+    token_file.write_text(json.dumps({"access_token": "rnk-token"}))
+
+    monkeypatch.setattr("renku_data_services.mcp_api.main._rnk_token_paths", lambda: [token_file])
+    assert _resolve_token() == "rnk-token"
+
+
+def test_resolve_token_raises_when_nothing_found(monkeypatch):
+    """_resolve_token raises RuntimeError with a helpful message when no token is available."""
+    monkeypatch.delenv("RENKU_ACCESS_TOKEN", raising=False)
+    monkeypatch.delenv("RENKU_TOKEN", raising=False)
+    monkeypatch.delenv("RENKU_CLI_ACCESS_TOKEN", raising=False)
+    monkeypatch.setattr("renku_data_services.mcp_api.main._rnk_token_paths", lambda: [])
+
+    with pytest.raises(RuntimeError, match="rnk login"):
+        _resolve_token()
+
+
+def test_rnk_token_paths_uses_xdg(monkeypatch):
+    """_rnk_token_paths respects XDG_DATA_HOME."""
+    monkeypatch.setenv("XDG_DATA_HOME", "/custom/xdg")
+    monkeypatch.delenv("APPDATA", raising=False)
+
+    paths = _rnk_token_paths()
+    assert any("/custom/xdg" in str(p) for p in paths)
+
+
 # ------------------------------------------------------------------ #
 # MCPDependencies.api — test via pytest-httpx                         #
 # ------------------------------------------------------------------ #