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
2 changes: 1 addition & 1 deletion packages/oauth/pyproject.toml
Original file line number Diff line number Diff line change
Expand Up @@ -9,8 +9,8 @@ authors = [{ name = "Keycard", email = "support@keycard.ai" }]
dependencies = [
"pydantic>=2.11.7",
"httpx>=0.28.1",
"authlib>=1.6.3",
"cryptography>=45.0.7",
"joserfc>=1.6.8",
]

keywords = ["oauth", "oauth2", "authentication", "tokens", "security"]
Expand Down
2 changes: 1 addition & 1 deletion packages/oauth/src/keycardai/oauth/server/__init__.py
Original file line number Diff line number Diff line change
@@ -1,7 +1,7 @@
"""Keycard OAuth Server Primitives.

Framework-free server components for protecting any HTTP API with Keycard.
These components depend only on pydantic, httpx, authlib, and cryptography —
These components depend only on pydantic, httpx, joserfc, and cryptography —
no MCP, Starlette, or other framework dependencies.

Core Components:
Expand Down
12 changes: 5 additions & 7 deletions packages/oauth/src/keycardai/oauth/server/private_key.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,11 @@
from pathlib import Path
from typing import Any, Protocol

from authlib.jose import JsonWebKey, JsonWebToken
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.serialization import PublicFormat
from joserfc import jwt as jose_jwt
from joserfc.jwk import import_key
from pydantic import AnyHttpUrl, BaseModel

from keycardai.oauth.types.models import (
Expand Down Expand Up @@ -209,7 +210,7 @@ def _generate_and_store_key_pair(self) -> None:
format=PublicFormat.SubjectPublicKeyInfo,
)

jwk = JsonWebKey.import_key(public_key_pem)
jwk = import_key(public_key_pem, "RSA")
public_key_jwk = jwk.as_dict()

public_key_jwk["kid"] = self.key_id
Expand Down Expand Up @@ -289,12 +290,9 @@ def create_client_assertion(

header = {"alg": "RS256", "typ": "JWT", "kid": self.key_id}

jwt = JsonWebToken(["RS256"])
private_key = serialization.load_pem_private_key(
self._private_key_pem.encode("utf-8"), password=None
)
private_key = import_key(self._private_key_pem, "RSA")

return jwt.encode(header, payload, private_key)
return jose_jwt.encode(header, payload, private_key)

def get_client_id(self) -> str:
return self.key_id
Expand Down
38 changes: 29 additions & 9 deletions packages/oauth/src/keycardai/oauth/utils/jwt.py
Original file line number Diff line number Diff line change
Expand Up @@ -34,14 +34,34 @@
import json
from typing import Any

from authlib.jose import JsonWebKey, JsonWebToken
from joserfc import jwt as jose_jwt
from joserfc.jwk import import_key
from pydantic import BaseModel

from ..exceptions import JWKSError, JWKSFetchError, JWKSKeyNotFoundError
from ..http._transports import HttpxAsyncTransport
from ..http._wire import HttpRequest
from ..types.models import ClientConfig

# joserfc requires an explicit key type when importing a PEM/DER key, otherwise
# it emits a SecurityWarning about implicit key types. Derive the key type from
# the JWS algorithm so PEM imports stay quiet and unambiguous. Importing from a
# JWK dict does not need this since the dict carries its own "kty".
_ALG_KEY_TYPE = {
"RS": "RSA",
"PS": "RSA",
"ES": "EC",
"Ed": "OKP",
"HS": "oct",
}


def _key_type_for_algorithm(algorithm: str) -> str:
key_type = _ALG_KEY_TYPE.get(algorithm[:2])
if key_type is None:
raise ValueError(f"Unsupported JWT algorithm: {algorithm}")
return key_type


def build_substitute_user_token(identifier: str) -> str:
"""Build an unsigned JWT for user impersonation via token exchange.
Expand Down Expand Up @@ -398,7 +418,7 @@ def get_scopes(self) -> list[str]:
def decode_and_verify_jwt(
jwt_token: str, verification_key: str, algorithm: str = "RS256"
) -> dict:
"""Decode and verify JWT token signature using authlib.
"""Decode and verify JWT token signature using joserfc.

Args:
jwt_token: JWT token string (without Bearer prefix)
Expand All @@ -412,9 +432,9 @@ def decode_and_verify_jwt(
ValueError: If token is invalid, malformed, or signature verification fails
"""
try:
jwt = JsonWebToken([algorithm])
claims = jwt.decode(jwt_token, verification_key)
return claims
key = import_key(verification_key, _key_type_for_algorithm(algorithm))
token = jose_jwt.decode(jwt_token, key, algorithms=[algorithm])
return token.claims
except Exception as e:
raise ValueError(f"JWT verification failed: {e}") from e

Expand Down Expand Up @@ -554,13 +574,13 @@ async def get_jwks_key(
if kid:
for key_data in keys:
if key_data.get("kid") == kid:
jwk = JsonWebKey.import_key(key_data)
return jwk.get_public_key() # type: ignore
jwk = import_key(key_data)
return jwk.as_pem().decode("utf-8")
raise JWKSKeyNotFoundError(f"Key ID '{kid}' not found")
else:
if len(keys) == 1:
jwk = JsonWebKey.import_key(keys[0])
return jwk.get_public_key() # type: ignore
jwk = import_key(keys[0])
return jwk.as_pem().decode("utf-8")
elif len(keys) > 1:
raise JWKSKeyNotFoundError("Multiple keys in JWKS but no key ID (kid) in token")
else:
Expand Down
62 changes: 34 additions & 28 deletions packages/oauth/tests/keycardai/oauth/utils/test_jwt.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,10 @@
from unittest.mock import AsyncMock, Mock, patch

import pytest
from authlib.jose import JsonWebKey, JsonWebSignature
from cryptography.hazmat.primitives import serialization
from cryptography.hazmat.primitives.asymmetric import rsa
from joserfc import jwt as jose_jwt
from joserfc.jwk import import_key

from keycardai.oauth.exceptions import JWKSFetchError, JWKSKeyNotFoundError
from keycardai.oauth.utils.jwt import (
Expand Down Expand Up @@ -321,25 +322,27 @@ def test_get_all_claims_optional_fields(self):
class TestJWTVerification:
"""Test JWT verification functionality."""

@patch("keycardai.oauth.utils.jwt.JsonWebToken")
def test_decode_and_verify_jwt_success(self, mock_jwt_class):
@patch("keycardai.oauth.utils.jwt.import_key")
@patch("keycardai.oauth.utils.jwt.jose_jwt")
def test_decode_and_verify_jwt_success(self, mock_jwt, mock_import_key):
"""Test successful JWT verification."""
mock_jwt = Mock()
mock_jwt.decode.return_value = {"sub": "user123", "iss": "example.com"}
mock_jwt_class.return_value = mock_jwt
mock_token = Mock()
mock_token.claims = {"sub": "user123", "iss": "example.com"}
mock_jwt.decode.return_value = mock_token

result = decode_and_verify_jwt("token", "key", "RS256")

assert result == {"sub": "user123", "iss": "example.com"}
mock_jwt_class.assert_called_once_with(["RS256"])
mock_jwt.decode.assert_called_once_with("token", "key")
mock_import_key.assert_called_once_with("key", "RSA")
mock_jwt.decode.assert_called_once_with(
"token", mock_import_key.return_value, algorithms=["RS256"]
)

@patch("keycardai.oauth.utils.jwt.JsonWebToken")
def test_decode_and_verify_jwt_failure(self, mock_jwt_class):
@patch("keycardai.oauth.utils.jwt.import_key")
@patch("keycardai.oauth.utils.jwt.jose_jwt")
def test_decode_and_verify_jwt_failure(self, mock_jwt, mock_import_key):
"""Test JWT verification failure."""
mock_jwt = Mock()
mock_jwt.decode.side_effect = Exception("Invalid signature")
mock_jwt_class.return_value = mock_jwt

with pytest.raises(ValueError, match="JWT verification failed"):
decode_and_verify_jwt("token", "key", "RS256")
Expand Down Expand Up @@ -433,9 +436,9 @@ async def test_get_verification_key_failure(self, mock_get_header):
@pytest.mark.asyncio
@patch("keycardai.oauth.utils.jwt.HttpxAsyncTransport")
@patch("keycardai.oauth.utils.jwt.ClientConfig")
@patch("keycardai.oauth.utils.jwt.JsonWebKey")
@patch("keycardai.oauth.utils.jwt.import_key")
async def test_get_jwks_key_with_kid(
self, mock_jwk_class, mock_config_class, mock_transport_class
self, mock_import_key, mock_config_class, mock_transport_class
):
"""Test JWKS key fetching with specific key ID."""
# Mock response
Expand All @@ -457,13 +460,13 @@ async def test_get_jwks_key_with_kid(

# Mock JWK
mock_jwk = Mock()
mock_jwk.get_public_key.return_value = "public_key_pem"
mock_jwk_class.import_key.return_value = mock_jwk
mock_jwk.as_pem.return_value = b"public_key_pem"
mock_import_key.return_value = mock_jwk

key = await get_jwks_key("key1", "https://example.com/.well-known/jwks.json")

assert key == "public_key_pem"
mock_jwk_class.import_key.assert_called_once_with(
mock_import_key.assert_called_once_with(
{"kid": "key1", "kty": "RSA", "use": "sig"}
)

Expand Down Expand Up @@ -507,9 +510,9 @@ async def test_get_jwks_key_http_error(
@pytest.mark.asyncio
@patch("keycardai.oauth.utils.jwt.HttpxAsyncTransport")
@patch("keycardai.oauth.utils.jwt.ClientConfig")
@patch("keycardai.oauth.utils.jwt.JsonWebKey")
@patch("keycardai.oauth.utils.jwt.import_key")
async def test_get_jwks_key_single_key_no_kid(
self, mock_jwk_class, mock_config_class, mock_transport_class
self, mock_import_key, mock_config_class, mock_transport_class
):
"""Test JWKS key fetching with single key and no kid parameter."""
mock_response = Mock()
Expand All @@ -523,13 +526,13 @@ async def test_get_jwks_key_single_key_no_kid(
mock_transport_class.return_value = mock_transport

mock_jwk = Mock()
mock_jwk.get_public_key.return_value = "public_key_pem"
mock_jwk_class.import_key.return_value = mock_jwk
mock_jwk.as_pem.return_value = b"public_key_pem"
mock_import_key.return_value = mock_jwk

key = await get_jwks_key(None, "https://example.com/.well-known/jwks.json")

assert key == "public_key_pem"
mock_jwk_class.import_key.assert_called_once_with({"kty": "RSA", "use": "sig"})
mock_import_key.assert_called_once_with({"kty": "RSA", "use": "sig"})

@pytest.mark.asyncio
@patch("keycardai.oauth.utils.jwt.HttpxAsyncTransport")
Expand Down Expand Up @@ -623,12 +626,8 @@ def create_token(claims: dict, kid: str = None, algorithm: str = "RS256") -> str

final_claims = {**default_claims, **claims}

jws = JsonWebSignature()
jwk = JsonWebKey.import_key(rsa_key_pair["private_pem"])

payload_json = json.dumps(final_claims)

token = jws.serialize_compact(header, payload_json, jwk)
private_key = import_key(rsa_key_pair["private_pem"], "RSA")
token = jose_jwt.encode(header, final_claims, private_key)
return token.decode('utf-8') if isinstance(token, bytes) else token

return create_token
Expand Down Expand Up @@ -706,6 +705,13 @@ def test_wrong_key_verification_failure(self, rsa_key_pair, jwt_token_factory):
with pytest.raises(ValueError, match="JWT verification failed"):
decode_and_verify_jwt(token, wrong_public_pem, "RS256")

def test_unsupported_algorithm_rejected(self, rsa_key_pair, jwt_token_factory):
"""Unknown algorithms are rejected, not silently mapped to a key type."""
token = jwt_token_factory({}, kid=rsa_key_pair["kid"])

with pytest.raises(ValueError, match="JWT verification failed"):
decode_and_verify_jwt(token, rsa_key_pair["public_pem"], "FOO256")


def test_parse_jwt_access_token_integration(self, rsa_key_pair, jwt_token_factory):
"""Test the complete parse_jwt_access_token flow with real crypto."""
Expand Down
10 changes: 5 additions & 5 deletions uv.lock

Some generated files are not rendered by default. Learn more about how customized files appear on GitHub.

Loading