@ -31,6 +31,7 @@ from __future__ import annotations
import json
import pathlib
from typing import Any
import aiofiles
from jwcrypto import jwk , jwt
@ -100,7 +101,7 @@ class KeycloakOpenID:
verify : bool | str = True ,
custom_headers : dict | None = None ,
proxies : dict | None = None ,
timeout : int = 60 ,
timeout : int | None = 60 ,
cert : str | tuple | None = None ,
max_retries : int = 1 ,
pool_maxsize : int | None = None ,
@ -166,7 +167,7 @@ class KeycloakOpenID:
self . _client_id = value
@property
def client_secret_key ( self ) - > str :
def client_secret_key ( self ) - > str | None :
"""
Get the client secret key .
@ -176,7 +177,7 @@ class KeycloakOpenID:
return self . _client_secret_key
@client_secret_key.setter
def client_secret_key ( self , value : str ) - > None :
def client_secret_key ( self , value : str | None ) - > None :
self . _client_secret_key = value
@property
@ -246,7 +247,7 @@ class KeycloakOpenID:
"""
return self . client_id + " / " + role
def _token_info ( self , token : str , method_token_info : str , * * kwargs : dict ) - > dict :
def _token_info ( self , token : str , method_token_info : str , * * kwargs : Any ) - > dict : # noqa: ANN401
"""
Getter for the token data .
@ -279,7 +280,15 @@ class KeycloakOpenID:
"""
params_path = { " realm-name " : self . realm_name }
data_raw = self . connection . raw_get ( URL_WELL_KNOWN . format ( * * params_path ) )
return raise_error_from_response ( data_raw , KeycloakGetError )
res = raise_error_from_response ( data_raw , KeycloakGetError )
if not isinstance ( res , dict ) :
msg = (
f " Unexpected response type on well_known. Expected ' dict ' , received ' {type(res)} ' "
f " , value: {res} "
)
raise TypeError ( msg )
return res
def auth_url (
self ,
@ -325,15 +334,15 @@ class KeycloakOpenID:
def token (
self ,
username : str = " " ,
password : str = " " ,
username : str | None = " " ,
password : str | None = " " ,
grant_type : str = " password " ,
code : str = " " ,
redirect_uri : str = " " ,
totp : int | None = None ,
scope : str = " openid " ,
code_verifier : str | None = None ,
* * extra : dict ,
* * extra : Any , # noqa: ANN401
) - > dict :
"""
Retrieve user token .
@ -385,7 +394,7 @@ class KeycloakOpenID:
payload [ " totp " ] = totp
payload = self . _add_secret_key ( payload )
content_type = self . connection . headers . get ( " Content-Type " )
content_type = ( self . connection . headers or { } ) . get ( " Content-Type " )
self . connection . add_param_headers ( " Content-Type " , " application/x-www-form-urlencoded " )
data_raw = self . connection . raw_post ( URL_TOKEN . format ( * * params_path ) , data = payload )
(
@ -393,7 +402,15 @@ class KeycloakOpenID:
if content_type
else self . connection . del_param_headers ( " Content-Type " )
)
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , dict ) :
msg = (
f " Unexpected response type from ' token ' . Expected ' dict ' , received ' {type(res)} ' "
f " , value {res}. "
)
raise TypeError ( msg )
return res
def refresh_token ( self , refresh_token : str , grant_type : str = " refresh_token " ) - > dict :
"""
@ -420,7 +437,7 @@ class KeycloakOpenID:
" refresh_token " : refresh_token ,
}
payload = self . _add_secret_key ( payload )
content_type = self . connection . headers . get ( " Content-Type " )
content_type = ( self . connection . headers or { } ) . get ( " Content-Type " )
self . connection . add_param_headers ( " Content-Type " , " application/x-www-form-urlencoded " )
data_raw = self . connection . raw_post ( URL_TOKEN . format ( * * params_path ) , data = payload )
(
@ -428,7 +445,16 @@ class KeycloakOpenID:
if content_type
else self . connection . del_param_headers ( " Content-Type " )
)
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type from refresh_token. "
f " Expected ' dict ' , received ' {type(res)} ' "
f " , value: {res}. "
)
raise TypeError ( msg )
return res
def exchange_token (
self ,
@ -480,7 +506,7 @@ class KeycloakOpenID:
" scope " : scope ,
}
payload = self . _add_secret_key ( payload )
content_type = self . connection . headers . get ( " Content-Type " )
content_type = ( self . connection . headers or { } ) . get ( " Content-Type " )
self . connection . add_param_headers ( " Content-Type " , " application/x-www-form-urlencoded " )
data_raw = self . connection . raw_post ( URL_TOKEN . format ( * * params_path ) , data = payload )
(
@ -488,7 +514,15 @@ class KeycloakOpenID:
if content_type
else self . connection . del_param_headers ( " Content-Type " )
)
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type from exchange_token. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' "
)
raise TypeError ( msg )
return res
def userinfo ( self , token : str ) - > dict :
"""
@ -504,7 +538,7 @@ class KeycloakOpenID:
: returns : Userinfo object
: rtype : dict
"""
orig_bearer = self . connection . headers . get ( " Authorization " )
orig_bearer = ( self . connection . headers or { } ) . get ( " Authorization " )
self . connection . add_param_headers ( " Authorization " , " Bearer " + token )
params_path = { " realm-name " : self . realm_name }
data_raw = self . connection . raw_get ( URL_USERINFO . format ( * * params_path ) )
@ -513,26 +547,42 @@ class KeycloakOpenID:
if orig_bearer is not None
else self . connection . del_param_headers ( " Authorization " )
)
return raise_error_from_response ( data_raw , KeycloakGetError )
res = raise_error_from_response ( data_raw , KeycloakGetError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type from userinfo. Expected ' dict ' , "
f " received ' {type(res)} ' , value: ' {res} ' . "
)
raise TypeError ( msg )
def logout ( self , refresh_token : str ) - > bytes :
return res
def logout ( self , refresh_token : str ) - > dict :
"""
Log out the authenticated user .
: param refresh_token : Refresh token from Keycloak
: type refresh_token : str
: returns : Keycloak server response
: rtype : bytes
: rtype : dict
"""
params_path = { " realm-name " : self . realm_name }
payload = { " client_id " : self . client_id , " refresh_token " : refresh_token }
payload = self . _add_secret_key ( payload )
data_raw = self . connection . raw_post ( URL_LOGOUT . format ( * * params_path ) , data = payload )
return raise_error_from_response (
res = raise_error_from_response (
data_raw ,
KeycloakPostError ,
expected_codes = [ HTTP_NO_CONTENT ] ,
)
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type from logout. Expected ' dict ' , "
f " received ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
def certs ( self ) - > dict :
"""
@ -549,7 +599,15 @@ class KeycloakOpenID:
"""
params_path = { " realm-name " : self . realm_name }
data_raw = self . connection . raw_get ( URL_CERTS . format ( * * params_path ) )
return raise_error_from_response ( data_raw , KeycloakGetError )
res = raise_error_from_response ( data_raw , KeycloakGetError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type from certs. Expected ' dict ' , "
f " received ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
def public_key ( self ) - > str :
"""
@ -562,7 +620,15 @@ class KeycloakOpenID:
"""
params_path = { " realm-name " : self . realm_name }
data_raw = self . connection . raw_get ( URL_REALM . format ( * * params_path ) )
return raise_error_from_response ( data_raw , KeycloakGetError ) [ " public_key " ]
res = raise_error_from_response ( data_raw , KeycloakGetError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type from public_key. Expected ' dict ' , "
f " received ' {type(res)} ' , value ' {res} ' "
)
raise TypeError ( msg )
return res [ " public_key " ]
def entitlement ( self , token : str , resource_server_id : str ) - > dict :
"""
@ -581,7 +647,7 @@ class KeycloakOpenID:
: returns : Entitlements
: rtype : dict
"""
orig_bearer = self . connection . headers . get ( " Authorization " )
orig_bearer = ( self . connection . headers or { } ) . get ( " Authorization " )
self . connection . add_param_headers ( " Authorization " , " Bearer " + token )
params_path = { " realm-name " : self . realm_name , " resource-server-id " : resource_server_id }
data_raw = self . connection . raw_get ( URL_ENTITLEMENT . format ( * * params_path ) )
@ -592,9 +658,24 @@ class KeycloakOpenID:
)
if data_raw . status_code in { HTTP_NOT_FOUND , HTTP_NOT_ALLOWED } :
return raise_error_from_response ( data_raw , KeycloakDeprecationError )
res = raise_error_from_response ( data_raw , KeycloakDeprecationError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , "
f " received ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
res = raise_error_from_response ( data_raw , KeycloakGetError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , "
f " received ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return raise_error_from_response ( data_raw , KeycloakGetError ) # pragma: no cover
return res
def introspect (
self ,
@ -629,7 +710,7 @@ class KeycloakOpenID:
if token_type_hint == " requesting_party_token " : # noqa: S105
if rpt :
payload . update ( { " token " : rpt , " token_type_hint " : token_type_hint } )
orig_bearer = self . connection . headers . get ( " Authorization " )
orig_bearer = ( self . connection . headers or { } ) . get ( " Authorization " )
self . connection . add_param_headers ( " Authorization " , " Bearer " + token )
bearer_changed = True
else :
@ -645,10 +726,19 @@ class KeycloakOpenID:
if orig_bearer is not None
else self . connection . del_param_headers ( " Authorization " )
)
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
@staticmethod
def _verify_token ( token : str , key : jwk . JWK | jwk . JWKSet | None , * * kwargs : dict ) - > dict :
def _verify_token ( token : str , key : jwk . JWK | jwk . JWKSet | None , * * kwargs : Any ) - > dict : # noqa: ANN401
"""
Decode and optionally validate a token .
@ -669,12 +759,13 @@ class KeycloakOpenID:
full_jwt = jwt . JWT ( jwt = token , * * kwargs )
full_jwt . leeway = leeway
full_jwt . validate ( key )
return jwt . json_decode ( full_jwt . claims )
return jwt . json_decode ( full_jwt . claims ) # pyright: ignore[reportAttributeAccessIssue]
full_jwt = jwt . JWT ( jwt = token , * * kwargs )
full_jwt . token . objects [ " valid " ] = True
return json . loads ( full_jwt . token . payload . decode ( " utf-8 " ) )
def decode_token ( self , token : str , validate : bool = True , * * kwargs : dict ) - > dict :
def decode_token ( self , token : str , validate : bool = True , * * kwargs : Any ) - > dict : # noqa: ANN401
"""
Decode user token .
@ -727,8 +818,8 @@ class KeycloakOpenID:
self ,
token : str ,
method_token_info : str = " introspect " , # noqa: S107
* * kwargs : dict ,
) - > list :
* * kwargs : Any , # noqa: ANN401
) - > list | None :
"""
Get policies by user token .
@ -771,8 +862,8 @@ class KeycloakOpenID:
self ,
token : str ,
method_token_info : str = " introspect " , # noqa: S107
* * kwargs : dict ,
) - > list :
* * kwargs : Any , # noqa: ANN401
) - > list | None :
"""
Get permission by user token .
@ -811,7 +902,7 @@ class KeycloakOpenID:
return list ( set ( permissions ) )
def uma_permissions ( self , token : str , permissions : str = " " , * * extra_payload : dict ) - > list :
def uma_permissions ( self , token : str , permissions : str = " " , * * extra_payload : Any ) - > list : # noqa: ANN401
"""
Get UMA permissions by user token with requested permissions .
@ -840,9 +931,9 @@ class KeycloakOpenID:
* * extra_payload ,
}
orig_bearer = self . connection . headers . get ( " Authorization " )
orig_bearer = ( self . connection . headers or { } ) . get ( " Authorization " )
self . connection . add_param_headers ( " Authorization " , " Bearer " + token )
content_type = self . connection . headers . get ( " Content-Type " )
content_type = ( self . connection . headers or { } ) . get ( " Content-Type " )
self . connection . add_param_headers ( " Content-Type " , " application/x-www-form-urlencoded " )
data_raw = self . connection . raw_post ( URL_TOKEN . format ( * * params_path ) , data = payload )
(
@ -855,9 +946,17 @@ class KeycloakOpenID:
if orig_bearer is not None
else self . connection . del_param_headers ( " Authorization " )
)
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , list ) :
msg = (
" Unexpected response type. Expected ' list ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
def has_uma_access ( self , token : str , permissions : list ) - > AuthStatus :
return res
def has_uma_access ( self , token : str , permissions : str ) - > AuthStatus :
"""
Determine whether user has uma permissions with specified user token .
@ -918,9 +1017,9 @@ class KeycloakOpenID:
: rtype : dict
"""
params_path = { " realm-name " : self . realm_name }
orig_bearer = self . connection . headers . get ( " Authorization " )
orig_bearer = ( self . connection . headers or { } ) . get ( " Authorization " )
self . connection . add_param_headers ( " Authorization " , " Bearer " + token )
orig_content_type = self . connection . headers . get ( " Content-Type " )
orig_content_type = ( self . connection . headers or { } ) . get ( " Content-Type " )
self . connection . add_param_headers ( " Content-Type " , " application/json " )
data_raw = self . connection . raw_post (
URL_CLIENT_REGISTRATION . format ( * * params_path ) ,
@ -936,7 +1035,15 @@ class KeycloakOpenID:
if orig_content_type is not None
else self . connection . del_param_headers ( " Content-Type " )
)
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
def device ( self , scope : str = " " ) - > dict :
"""
@ -964,9 +1071,17 @@ class KeycloakOpenID:
payload = self . _add_secret_key ( payload )
data_raw = self . connection . raw_post ( URL_DEVICE . format ( * * params_path ) , data = payload )
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
def update_client ( self , token : str , client_id : str , payload : dict ) - > bytes :
def update_client ( self , token : str , client_id : str , payload : dict ) - > dict :
"""
Update a client .
@ -983,9 +1098,9 @@ class KeycloakOpenID:
: rtype : bytes
"""
params_path = { " realm-name " : self . realm_name , " client-id " : client_id }
orig_bearer = self . connection . headers . get ( " Authorization " )
orig_bearer = ( self . connection . headers or { } ) . get ( " Authorization " )
self . connection . add_param_headers ( " Authorization " , " Bearer " + token )
orig_content_type = self . connection . headers . get ( " Content-Type " )
orig_content_type = ( self . connection . headers or { } ) . get ( " Content-Type " )
self . connection . add_param_headers ( " Content-Type " , " application/json " )
# Keycloak complains if the clientId is not set in the payload
@ -1006,9 +1121,17 @@ class KeycloakOpenID:
if orig_content_type is not None
else self . connection . del_param_headers ( " Content-Type " )
)
return raise_error_from_response ( data_raw , KeycloakPutError )
res = raise_error_from_response ( data_raw , KeycloakPutError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
async def _a_token_info ( self , token : str , method_token_info : str , * * kwargs : dict ) - > dict :
async def _a_token_info ( self , token : str , method_token_info : str , * * kwargs : Any ) - > dict : # noqa: ANN401
"""
Asynchronous getter for the token data .
@ -1041,7 +1164,15 @@ class KeycloakOpenID:
"""
params_path = { " realm-name " : self . realm_name }
data_raw = await self . connection . a_raw_get ( URL_WELL_KNOWN . format ( * * params_path ) )
return raise_error_from_response ( data_raw , KeycloakGetError )
res = raise_error_from_response ( data_raw , KeycloakGetError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
async def a_auth_url (
self ,
@ -1087,15 +1218,15 @@ class KeycloakOpenID:
async def a_token (
self ,
username : str = " " ,
password : str = " " ,
username : str | None = " " ,
password : str | None = " " ,
grant_type : str = " password " ,
code : str = " " ,
redirect_uri : str = " " ,
totp : int | None = None ,
scope : str = " openid " ,
code_verifier : str | None = None ,
* * extra : dict ,
* * extra : Any , # noqa: ANN401
) - > dict :
"""
Retrieve user token asynchronously .
@ -1147,7 +1278,7 @@ class KeycloakOpenID:
payload [ " totp " ] = totp
payload = self . _add_secret_key ( payload )
content_type = self . connection . headers . get ( " Content-Type " )
content_type = ( self . connection . headers or { } ) . get ( " Content-Type " )
self . connection . add_param_headers ( " Content-Type " , " application/x-www-form-urlencoded " )
data_raw = await self . connection . a_raw_post ( URL_TOKEN . format ( * * params_path ) , data = payload )
(
@ -1155,7 +1286,15 @@ class KeycloakOpenID:
if content_type
else self . connection . del_param_headers ( " Content-Type " )
)
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
async def a_refresh_token ( self , refresh_token : str , grant_type : str = " refresh_token " ) - > dict :
"""
@ -1182,7 +1321,7 @@ class KeycloakOpenID:
" refresh_token " : refresh_token ,
}
payload = self . _add_secret_key ( payload )
content_type = self . connection . headers . get ( " Content-Type " )
content_type = ( self . connection . headers or { } ) . get ( " Content-Type " )
self . connection . add_param_headers ( " Content-Type " , " application/x-www-form-urlencoded " )
data_raw = await self . connection . a_raw_post ( URL_TOKEN . format ( * * params_path ) , data = payload )
(
@ -1190,7 +1329,15 @@ class KeycloakOpenID:
if content_type
else self . connection . del_param_headers ( " Content-Type " )
)
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
async def a_exchange_token (
self ,
@ -1242,7 +1389,7 @@ class KeycloakOpenID:
" scope " : scope ,
}
payload = self . _add_secret_key ( payload )
content_type = self . connection . headers . get ( " Content-Type " )
content_type = ( self . connection . headers or { } ) . get ( " Content-Type " )
self . connection . add_param_headers ( " Content-Type " , " application/x-www-form-urlencoded " )
data_raw = await self . connection . a_raw_post ( URL_TOKEN . format ( * * params_path ) , data = payload )
(
@ -1250,7 +1397,15 @@ class KeycloakOpenID:
if content_type
else self . connection . del_param_headers ( " Content-Type " )
)
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
async def a_userinfo ( self , token : str ) - > dict :
"""
@ -1266,7 +1421,7 @@ class KeycloakOpenID:
: returns : Userinfo object
: rtype : dict
"""
orig_bearer = self . connection . headers . get ( " Authorization " )
orig_bearer = ( self . connection . headers or { } ) . get ( " Authorization " )
self . connection . add_param_headers ( " Authorization " , " Bearer " + token )
params_path = { " realm-name " : self . realm_name }
data_raw = await self . connection . a_raw_get ( URL_USERINFO . format ( * * params_path ) )
@ -1275,26 +1430,42 @@ class KeycloakOpenID:
if orig_bearer is not None
else self . connection . del_param_headers ( " Authorization " )
)
return raise_error_from_response ( data_raw , KeycloakGetError )
res = raise_error_from_response ( data_raw , KeycloakGetError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
async def a_logout ( self , refresh_token : str ) - > bytes :
return res
async def a_logout ( self , refresh_token : str ) - > dict :
"""
Log out the authenticated user asynchronously .
: param refresh_token : Refresh token from Keycloak
: type refresh_token : str
: returns : Keycloak server response
: rtype : bytes
: rtype : dict
"""
params_path = { " realm-name " : self . realm_name }
payload = { " client_id " : self . client_id , " refresh_token " : refresh_token }
payload = self . _add_secret_key ( payload )
data_raw = await self . connection . a_raw_post ( URL_LOGOUT . format ( * * params_path ) , data = payload )
return raise_error_from_response (
res = raise_error_from_response (
data_raw ,
KeycloakPostError ,
expected_codes = [ HTTP_NO_CONTENT ] ,
)
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
async def a_certs ( self ) - > dict :
"""
@ -1311,7 +1482,15 @@ class KeycloakOpenID:
"""
params_path = { " realm-name " : self . realm_name }
data_raw = await self . connection . a_raw_get ( URL_CERTS . format ( * * params_path ) )
return raise_error_from_response ( data_raw , KeycloakGetError )
res = raise_error_from_response ( data_raw , KeycloakGetError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
async def a_public_key ( self ) - > str :
"""
@ -1324,7 +1503,15 @@ class KeycloakOpenID:
"""
params_path = { " realm-name " : self . realm_name }
data_raw = await self . connection . a_raw_get ( URL_REALM . format ( * * params_path ) )
return raise_error_from_response ( data_raw , KeycloakGetError ) [ " public_key " ]
res = raise_error_from_response ( data_raw , KeycloakGetError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res [ " public_key " ]
async def a_entitlement ( self , token : str , resource_server_id : str ) - > dict :
"""
@ -1343,7 +1530,7 @@ class KeycloakOpenID:
: returns : Entitlements
: rtype : dict
"""
orig_bearer = self . connection . headers . get ( " Authorization " )
orig_bearer = ( self . connection . headers or { } ) . get ( " Authorization " )
self . connection . add_param_headers ( " Authorization " , " Bearer " + token )
params_path = { " realm-name " : self . realm_name , " resource-server-id " : resource_server_id }
data_raw = await self . connection . a_raw_get ( URL_ENTITLEMENT . format ( * * params_path ) )
@ -1354,9 +1541,25 @@ class KeycloakOpenID:
)
if data_raw . status_code in [ HTTP_NOT_FOUND , HTTP_NOT_ALLOWED ] :
return raise_error_from_response ( data_raw , KeycloakDeprecationError )
res = raise_error_from_response ( data_raw , KeycloakDeprecationError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
return raise_error_from_response ( data_raw , KeycloakGetError ) # pragma: no cover
res = raise_error_from_response ( data_raw , KeycloakGetError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
async def a_introspect (
self ,
@ -1391,7 +1594,7 @@ class KeycloakOpenID:
if token_type_hint == " requesting_party_token " : # noqa: S105
if rpt :
payload . update ( { " token " : rpt , " token_type_hint " : token_type_hint } )
orig_bearer = self . connection . headers . get ( " Authorization " )
orig_bearer = ( self . connection . headers or { } ) . get ( " Authorization " )
self . connection . add_param_headers ( " Authorization " , " Bearer " + token )
bearer_changed = True
else :
@ -1410,9 +1613,17 @@ class KeycloakOpenID:
if orig_bearer is not None
else self . connection . del_param_headers ( " Authorization " )
)
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
async def a_decode_token ( self , token : str , validate : bool = True , * * kwargs : dict ) - > dict :
return res
async def a_decode_token ( self , token : str , validate : bool = True , * * kwargs : Any ) - > dict : # noqa: ANN401
"""
Decode user token asynchronously .
@ -1465,8 +1676,8 @@ class KeycloakOpenID:
self ,
token : str ,
method_token_info : str = " introspect " , # noqa: S107
* * kwargs : dict ,
) - > list :
* * kwargs : Any , # noqa: ANN401
) - > list | None :
"""
Get policies by user token asynchronously .
@ -1477,7 +1688,7 @@ class KeycloakOpenID:
: param kwargs : Additional keyword arguments
: type kwargs : dict
: return : Policies
: rtype : list
: rtype : list | None
: raises KeycloakAuthorizationConfigError : In case of bad authorization configuration
: raises KeycloakInvalidTokenError : In case of bad token
"""
@ -1509,8 +1720,8 @@ class KeycloakOpenID:
self ,
token : str ,
method_token_info : str = " introspect " , # noqa: S107
* * kwargs : dict ,
) - > list :
* * kwargs : Any , # noqa: ANN401
) - > list | None :
"""
Get permission by user token asynchronously .
@ -1521,7 +1732,7 @@ class KeycloakOpenID:
: param kwargs : parameters for decode
: type kwargs : dict
: returns : permissions list
: rtype : list
: rtype : list | None
: raises KeycloakAuthorizationConfigError : In case of bad authorization configuration
: raises KeycloakInvalidTokenError : In case of bad token
"""
@ -1552,7 +1763,7 @@ class KeycloakOpenID:
self ,
token : str ,
permissions : str = " " ,
* * extra_payload : dict ,
* * extra_payload : Any , # noqa: ANN401
) - > list :
"""
Get UMA permissions by user token with requested permissions asynchronously .
@ -1582,9 +1793,9 @@ class KeycloakOpenID:
* * extra_payload ,
}
orig_bearer = self . connection . headers . get ( " Authorization " )
orig_bearer = ( self . connection . headers or { } ) . get ( " Authorization " )
self . connection . add_param_headers ( " Authorization " , " Bearer " + token )
content_type = self . connection . headers . get ( " Content-Type " )
content_type = ( self . connection . headers or { } ) . get ( " Content-Type " )
self . connection . add_param_headers ( " Content-Type " , " application/x-www-form-urlencoded " )
data_raw = await self . connection . a_raw_post ( URL_TOKEN . format ( * * params_path ) , data = payload )
(
@ -1597,9 +1808,17 @@ class KeycloakOpenID:
if orig_bearer is not None
else self . connection . del_param_headers ( " Authorization " )
)
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , list ) :
msg = (
" Unexpected response type. Expected ' list ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
async def a_has_uma_access ( self , token : str , permissions : list ) - > AuthStatus :
return res
async def a_has_uma_access ( self , token : str , permissions : str ) - > AuthStatus :
"""
Determine whether user has uma permissions with specified user token asynchronously .
@ -1660,9 +1879,9 @@ class KeycloakOpenID:
: rtype : dict
"""
params_path = { " realm-name " : self . realm_name }
orig_bearer = self . connection . headers . get ( " Authorization " )
orig_bearer = ( self . connection . headers or { } ) . get ( " Authorization " )
self . connection . add_param_headers ( " Authorization " , " Bearer " + token )
orig_content_type = self . connection . headers . get ( " Content-Type " )
orig_content_type = ( self . connection . headers or { } ) . get ( " Content-Type " )
self . connection . add_param_headers ( " Content-Type " , " application/json " )
data_raw = await self . connection . a_raw_post (
URL_CLIENT_REGISTRATION . format ( * * params_path ) ,
@ -1678,7 +1897,15 @@ class KeycloakOpenID:
if orig_content_type is not None
else self . connection . del_param_headers ( " Content-Type " )
)
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
async def a_device ( self , scope : str = " " ) - > dict :
"""
@ -1706,9 +1933,17 @@ class KeycloakOpenID:
payload = self . _add_secret_key ( payload )
data_raw = await self . connection . a_raw_post ( URL_DEVICE . format ( * * params_path ) , data = payload )
return raise_error_from_response ( data_raw , KeycloakPostError )
res = raise_error_from_response ( data_raw , KeycloakPostError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
async def a_update_client ( self , token : str , client_id : str , payload : dict ) - > bytes :
async def a_update_client ( self , token : str , client_id : str , payload : dict ) - > dict :
"""
Update a client asynchronously .
@ -1722,12 +1957,12 @@ class KeycloakOpenID:
: param payload : ClientRepresentation
: type payload : dict
: return : Client Representation
: rtype : bytes
: rtype : dict
"""
params_path = { " realm-name " : self . realm_name , " client-id " : client_id }
orig_bearer = self . connection . headers . get ( " Authorization " )
orig_bearer = ( self . connection . headers or { } ) . get ( " Authorization " )
self . connection . add_param_headers ( " Authorization " , " Bearer " + token )
orig_content_type = self . connection . headers . get ( " Content-Type " )
orig_content_type = ( self . connection . headers or { } ) . get ( " Content-Type " )
self . connection . add_param_headers ( " Content-Type " , " application/json " )
# Keycloak complains if the clientId is not set in the payload
@ -1748,4 +1983,12 @@ class KeycloakOpenID:
if orig_content_type is not None
else self . connection . del_param_headers ( " Content-Type " )
)
return raise_error_from_response ( data_raw , KeycloakPutError )
res = raise_error_from_response ( data_raw , KeycloakPutError )
if not isinstance ( res , dict ) :
msg = (
" Unexpected response type. Expected ' dict ' , received "
f " ' {type(res)} ' , value ' {res} ' . "
)
raise TypeError ( msg )
return res
Richard Nemeth
2 months ago