Merge pull request #30 from keycardai/revert-29-feat/2-step-eks-workload-identity

Revert "feat(keycardai-mcp): app credential grant flow"

Author: kamil-keycard

packages/mcp-fastmcp/pyproject.toml

@@ -12,7 +12,7 @@ dependencies = [
     "httpx>=0.27.2",
     "keycardai-oauth>=0.5.0",
     "fastmcp==2.12.0",
-    "keycardai-mcp>=0.10.0",
+    "keycardai-mcp>=0.11.0",
 ]
 keywords = ["fastmcp", "mcp", "model-context-protocol", "oauth", "token-exchange", "authentication", "keycard"]
 classifiers = [

packages/mcp/pyproject.toml

@@ -13,7 +13,6 @@ dependencies = [
     "httpx>=0.27.2",
     "starlette>=0.47.3",
     "nanoid>=2.0.0",
-    "pyjwt>=2.10.1",
 ]
 keywords = ["mcp", "model-context-protocol", "authentication", "authorization", "ai", "llm"]
 classifiers = [

packages/mcp/src/keycardai/mcp/server/auth/_cache.py

@@ -3,7 +3,7 @@
 import threading
 import time
 from dataclasses import dataclass
-from typing import Any, Protocol
+from typing import Any
 
 
 @dataclass
@@ -155,40 +155,3 @@ def cleanup_expired(self) -> int:
                 del self._cache[cache_key]
 
             return len(expired_keys)
-
-class TokenCache(Protocol):
-    """Protocol for token cache implementations."""
-
-    def get(self, key: str) -> tuple[str, int] | None:
-        """Get a value from the cache if it exists and hasn't expired."""
-        pass
-
-    def set(self, key: str, value: tuple[str, int]) -> None:
-        """Set a value in the cache with current timestamp."""
-        pass
-
-class InMemoryTokenCache(TokenCache):
-    """In-memory token cache implementation."""
-
-    def __init__(self, exp_leeway: int = 300):
-        self.exp_leeway = exp_leeway
-        self._cache: dict[str, tuple[str, int]] = {}
-
-    def get(self, key: str) -> tuple[str, int] | None:
-        cached = self._cache.get(key)
-        if not cached:
-            return None
-
-        access_token, exp_time = cached
-        # Check if token is expired with leeway (default 5 minutes before exp)
-        # exp_time is epoch timestamp from JWT 'exp' claim
-        current_time = int(time.time())
-        if current_time >= (exp_time - self.exp_leeway):
-            # Token expired or too close to expiration - force refresh
-            self._cache.pop(key)
-            return None
-
-        return cached
-
-    def set(self, key: str, value: tuple[str, int]) -> None:
-        self._cache[key] = value

packages/mcp/src/keycardai/mcp/server/auth/application_credentials.py

@@ -18,11 +18,8 @@
 
 import os
 import uuid
-from asyncio import Lock
 from typing import Protocol
 
-from jwt import decode
-
 from keycardai.oauth import (
     AsyncClient,
     AuthStrategy,
@@ -39,7 +36,6 @@
     EKSWorkloadIdentityConfigurationError,
     EKSWorkloadIdentityRuntimeError,
 )
-from ._cache import InMemoryTokenCache, TokenCache
 from .private_key import (
     FilePrivateKeyStorage,
     PrivateKeyManager,
@@ -432,7 +428,6 @@ def __init__(
         self,
         token_file_path: str | None = None,
         env_var_name: str = "AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE",
-        cache: TokenCache | None = None
     ):
         """Initialize EKS workload identity provider.
 
@@ -441,16 +436,10 @@ def __init__(
                            reads from the environment variable specified by env_var_name.
             env_var_name: Name of the environment variable containing the token file path.
                          Defaults to AWS_CONTAINER_AUTHORIZATION_TOKEN_FILE.
-            cache_leeway: Number of seconds before JWT expiration to consider it expired.
-                         Defaults to 300 seconds (5 minutes) for safety margin.
 
         Raises:
             EKSWorkloadIdentityConfigurationError: If token file cannot be read or is empty.
         """
-        if cache is None:
-            cache = InMemoryTokenCache()
-        self._application_credentials: TokenCache = cache  # jti -> (access_token, exp)
-        self._lock = Lock()
         self.env_var_name = env_var_name
 
         if token_file_path is not None:
@@ -570,107 +559,6 @@ def set_client_config(
         """
         return config
 
-    def _get_application_credential(self, client_assertion: str) -> tuple[str, int] | None:
-        """Get cached application credential if valid and not expired.
-
-        Uses double-checked locking pattern - this is the fast path that checks
-        cache WITHOUT acquiring lock for maximum performance.
-
-        Args:
-            client_assertion: The EKS workload identity JWT token
-
-        Returns:
-            Tuple of (access_token, exp) if cached and valid, None if expired or not cached
-        """
-        try:
-            # Decode without verification - we just need the jti claim for cache lookup
-            # The Keycard backend validates the actual assertion signature
-            decoded_assertion = decode(client_assertion, options={"verify_signature": False})
-        except Exception:
-            return None
-
-        assertion_jti = decoded_assertion.get("jti")
-        if not assertion_jti:
-            return None
-
-        return self._application_credentials.get(assertion_jti)
-
-    def _set_application_credential(self, client_assertion: str, access_token: str) -> None:
-        """Cache the application credential with expiration from JWT.
-
-        Extracts the 'exp' claim from the access token JWT and stores it
-        for expiration checking. This method is called inside the lock.
-
-        Args:
-            client_assertion: The EKS workload identity JWT (used for cache key)
-            access_token: The access token JWT (contains 'exp' claim)
-        """
-        try:
-            decoded_assertion = decode(client_assertion, options={"verify_signature": False})
-        except Exception:
-            return
-
-        assertion_jti = decoded_assertion.get("jti")
-        if not assertion_jti:
-            return
-
-        try:
-            # Decode access token to get exp claim for expiration checking
-            decoded_token = decode(access_token, options={"verify_signature": False})
-        except Exception:
-            return
-
-        exp_time = decoded_token.get("exp")
-        if not exp_time:
-            return
-
-        # Cache with epoch expiration time from JWT
-        self._application_credentials.set(assertion_jti, (access_token, exp_time))
-
-    async def get_application_credential(self, client: AsyncClient, client_assertion: str) -> str:
-        """Get application credential.
-
-        Args:
-            client: OAuth client for token exchange
-            client_assertion: The EKS workload identity JWT token
-
-        Returns:
-            The access token for the application credential
-
-        Raises:
-            EKSWorkloadIdentityRuntimeError: If token exchange fails
-        """
-        cached = self._get_application_credential(client_assertion)
-        if cached:
-            access_token, _ = cached
-            return access_token
-
-        async with self._lock:
-            # DOUBLE CHECK: Another coroutine may have just populated cache while we waited
-            cached = self._get_application_credential(client_assertion)
-            if cached:
-                access_token, _ = cached
-                return access_token
-
-            request = TokenExchangeRequest(
-                grant_type=GrantType.CLIENT_CREDENTIALS,
-                client_assertion_type=GrantType.JWT_BEARER_CLIENT_ASSERTION,
-                client_assertion=client_assertion
-            )
-
-            try:
-                response = await client.exchange_token(request)
-            except Exception as e:
-                raise EKSWorkloadIdentityRuntimeError(
-                    token_file_path=self.token_file_path,
-                    env_var_name=self.env_var_name,
-                    error_details=f"Error getting application credential: {str(e)}",
-                ) from e
-
-            self._set_application_credential(client_assertion, response.access_token)
-            return response.access_token
-
-
     async def prepare_token_exchange_request(
         self,
         client: AsyncClient,
@@ -699,13 +587,11 @@ async def prepare_token_exchange_request(
         # Read the token from the filesystem
         eks_token = self._read_token()
 
-        application_credential = await self.get_application_credential(client, eks_token)
-
         return TokenExchangeRequest(
             subject_token=subject_token,
             resource=resource,
             subject_token_type="urn:ietf:params:oauth:token-type:access_token",
             client_assertion_type=GrantType.JWT_BEARER_CLIENT_ASSERTION,
-            client_assertion=application_credential,
+            client_assertion=eks_token,
         )