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
10 changes: 10 additions & 0 deletions packages/oauth/src/keycardai/oauth/client.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -814,6 +814,7 @@ async def client_credentials_grant(
scope: str | None = None,
client_assertion: str | None = None,
client_assertion_type: str | None = None,
client_id: str | None = None,
timeout: float | None = None,
) -> TokenResponse: ...

Expand DownExpand Up@@ -842,6 +843,10 @@ async def client_credentials_grant(self, request: ClientCredentialsRequest | Non
strategy (e.g. MultiZoneBasicAuth) applies for this call.
Defaults to the client's issuer. Single-credential strategies
ignore it. May be combined with either calling form.
client_id: Client identifier sent in the request body. Required when a
federation-rule workload identity credential presents a jwt-bearer
client_assertion and the zone resolves the application credential by
application ID.
**client_credentials_args: Alternative to request - provide individual parameters

Returns:
Expand DownExpand Up@@ -1544,6 +1549,7 @@ def client_credentials_grant(
scope: str | None = None,
client_assertion: str | None = None,
client_assertion_type: str | None = None,
client_id: str | None = None,
timeout: float | None = None,
) -> TokenResponse: ...

Expand DownExpand Up@@ -1572,6 +1578,10 @@ def client_credentials_grant(self, request: ClientCredentialsRequest | None = No
strategy (e.g. MultiZoneBasicAuth) applies for this call.
Defaults to the client's issuer. Single-credential strategies
ignore it. May be combined with either calling form.
client_id: Client identifier sent in the request body. Required when a
federation-rule workload identity credential presents a jwt-bearer
client_assertion and the zone resolves the application credential by
application ID.
**client_credentials_args: Alternative to request - provide individual parameters

Returns:
Expand Down
9 changes: 9 additions & 0 deletions packages/oauth/src/keycardai/oauth/types/models.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -64,6 +64,15 @@ class ClientCredentialsRequest(BaseModel):
scope: str | None = Field(default=None, description="Space-delimited scope of the access request.")
client_assertion: str | None = None
client_assertion_type: str | None = None
client_id: str | None = Field(
default=None,
description=(
"The client identifier. Accompanies a jwt-bearer client_assertion when the "
"zone resolves the application credential by application ID rather than by "
"the assertion subject. Not needed when the client authenticates at the HTTP "
"layer, such as Basic auth."
),
)
timeout: float | None = None


Expand Down
Original file line numberDiff line numberDiff line change
Expand Up@@ -57,6 +57,39 @@ def test_build_client_credentials_http_request_full(self):
"client_assertion_type": ["urn:ietf:params:oauth:client-assertion-type:jwt-bearer"],
}

def test_client_credentials_request_client_id_defaults_to_none(self):
"""Test client_id defaults to None on the request model."""
assert ClientCredentialsRequest().client_id is None
assert ClientCredentialsRequest(client_id="app_123").client_id == "app_123"

def test_build_client_credentials_http_request_with_client_id(self):
"""Test client_id is encoded in the form body when set."""
req = ClientCredentialsRequest(
client_id="app_123",
client_assertion="assertion_jwt",
client_assertion_type="urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
)

http_req = build_client_credentials_http_request(req, HTTPContext(endpoint="https://auth.example.com/token", transport=Mock(), auth=NoneAuth()))

form = parse_qs(http_req.body.decode("utf-8"))
assert form == {
"grant_type": ["client_credentials"],
"client_id": ["app_123"],
"client_assertion": ["assertion_jwt"],
"client_assertion_type": ["urn:ietf:params:oauth:client-assertion-type:jwt-bearer"],
}

def test_build_client_credentials_http_request_omits_unset_client_id(self):
"""Test the client_id key is absent from the form body when unset."""
req = ClientCredentialsRequest(scope="read")

http_req = build_client_credentials_http_request(req, HTTPContext(endpoint="https://auth.example.com/token", transport=Mock(), auth=NoneAuth()))

form = parse_qs(http_req.body.decode("utf-8"))
assert "client_id" not in form
assert form == {"grant_type": ["client_credentials"], "scope": ["read"]}

def test_parse_client_credentials_http_response_success(self):
"""Test parsing successful client credentials response."""
response_body = b'''{
Expand DownExpand Up@@ -204,3 +237,64 @@ async def test_client_credentials_grant_async(self):
"grant_type": ["client_credentials"],
"resource": ["https://api.example.com"],
}

def test_client_credentials_grant_sync_sends_client_id(self):
"""Test the sync grant sends client_id alongside a jwt-bearer assertion."""
mock_transport = Mock()
mock_transport.request_raw.return_value = HttpResponse(
status=200,
headers={"Content-Type": "application/json"},
body=b'{"access_token": "sync_issued_token", "token_type": "Bearer"}'
)

context = HTTPContext(
endpoint="https://auth.example.com/token",
transport=mock_transport,
auth=NoneAuth(),
timeout=30.0
)

req = ClientCredentialsRequest(
client_id="app_123",
client_assertion="assertion_jwt",
client_assertion_type="urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
)

result = client_credentials_grant(req, context)

assert result.access_token == "sync_issued_token"

sent_request = mock_transport.request_raw.call_args[0][0]
form = parse_qs(sent_request.body.decode("utf-8"))
assert form["client_id"] == ["app_123"]

@pytest.mark.asyncio
async def test_client_credentials_grant_async_sends_client_id(self):
"""Test the async grant sends client_id alongside a jwt-bearer assertion."""
mock_transport = AsyncMock()
mock_transport.request_raw.return_value = HttpResponse(
status=200,
headers={"Content-Type": "application/json"},
body=b'{"access_token": "async_issued_token", "token_type": "Bearer"}'
)

context = HTTPContext(
endpoint="https://auth.example.com/token",
transport=mock_transport,
auth=NoneAuth(),
timeout=30.0
)

req = ClientCredentialsRequest(
client_id="app_123",
client_assertion="assertion_jwt",
client_assertion_type="urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
)

result = await client_credentials_grant_async(req, context)

assert result.access_token == "async_issued_token"

sent_request = mock_transport.request_raw.call_args[0][0]
form = parse_qs(sent_request.body.decode("utf-8"))
assert form["client_id"] == ["app_123"]
31 changes: 31 additions & 0 deletions packages/oauth/tests/keycardai/oauth/test_client.py
Original file line numberDiff line numberDiff line change
Expand Up@@ -400,6 +400,37 @@ async def test_async_client_credentials_grant_overload_equivalence(self):

assert dict1 == dict2, f"Async client credentials requests differ: {dict1} != {dict2}"

def test_client_credentials_grant_kwargs_passes_client_id(self):
"""Test the sync kwargs form passes client_id into the request."""
with patch('keycardai.oauth.client.client_credentials_grant') as mock_grant:
mock_grant.return_value = Mock()
client = Client("https://test.keycard.cloud")

client.client_credentials_grant(
client_id="app_123",
client_assertion="assertion_jwt",
client_assertion_type="urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
)

request = mock_grant.call_args[0][0]
assert request.client_id == "app_123"

@pytest.mark.asyncio
async def test_async_client_credentials_grant_kwargs_passes_client_id(self):
"""Test the async kwargs form passes client_id into the request."""
with patch('keycardai.oauth.client.client_credentials_grant_async') as mock_grant_async:
mock_grant_async.return_value = Mock()
async_client = AsyncClient("https://test.keycard.cloud")

await async_client.client_credentials_grant(
client_id="app_123",
client_assertion="assertion_jwt",
client_assertion_type="urn:ietf:params:oauth:client-assertion-type:jwt-bearer",
)

request = mock_grant_async.call_args[0][0]
assert request.client_id == "app_123"

def test_discover_server_metadata_overload_equivalence(self):
"""Test that discover_server_metadata overloads create equivalent calls."""
test_base_url = "https://custom.auth.server.com"
Expand Down
Loading