keycloak.keycloak_uma¶

Keycloak UMA module.

The module contains a UMA compatible client for keycloak: https://docs.kantarainitiative.org/uma/wg/rec-oauth-uma-federated-authz-2.0.html

Attributes¶

URL_UMA_WELL_KNOWN

Exceptions¶

`KeycloakDeleteError`	Keycloak request delete error exception.
`KeycloakGetError`	Keycloak request get error exception.
`KeycloakPostError`	Keycloak request post error exception.
`KeycloakPutError`	Keycloak request put error exception.

Classes¶

`ConnectionManager`	Represents a simple server connection.
`KeycloakOpenIDConnection`	A class to help with OpenID connections which can auto refresh tokens.
`UMAPermission`	A class to conveniently assemble permissions.
`KeycloakUMA`	Keycloak UMA client.

Functions¶

raise_error_from_response(response, error[, ...])

Raise an exception for the response.

Module Contents¶

class keycloak.keycloak_uma.ConnectionManager(base_url, headers={}, timeout=60, verify=True, proxies=None)[source]¶

Bases: object

Represents a simple server connection.

Parameters:

base_url (str) – The server URL.
headers (dict) – The header parameters of the requests to the server.
timeout (int) – Timeout to use for requests to the server.
verify (Union[bool,str]) – Boolean value to enable or disable certificate validation or a string containing a path to a CA bundle to use
proxies (dict) – The proxies servers requests is sent by.

async aclose()[source]¶: Close the async connection on delete.

__del__()[source]¶: Del method.

property base_url¶

Return base url in use for requests to the server.

Returns:: Base URL
Return type:: str

property timeout¶

Return timeout in use for request to the server.

Returns:: Timeout
Return type:: int

property verify¶

Return verify in use for request to the server.

Returns:: Verify indicator
Return type:: bool

property headers¶

Return header request to the server.

Returns:: Request headers
Return type:: dict

param_headers(key)[source]¶

Return a specific header parameter.

Parameters:: key (str) – Header parameters key.
Returns:: If the header parameters exist, return its value.
Return type:: str

clean_headers()[source]¶: Clear header parameters.

exist_param_headers(key)[source]¶

Check if the parameter exists in the header.

Parameters:: key (str) – Header parameters key.
Returns:: If the header parameters exist, return True.
Return type:: bool

add_param_headers(key, value)[source]¶

Add a single parameter inside the header.

Parameters:

key (str) – Header parameters key.
value (str) – Value to be added.

del_param_headers(key)[source]¶

Remove a specific parameter.

Parameters:: key (str) – Key of the header parameters.

raw_get(path, **kwargs)[source]¶

Submit get request to the path.

Parameters:

path (str) – Path for request.
kwargs (dict) – Additional arguments

Returns:

Response the request.

Return type:

Response

Raises:

KeycloakConnectionError – HttpError Can’t connect to server.

raw_post(path, data, **kwargs)[source]¶

Submit post request to the path.

Parameters:

path (str) – Path for request.
data (dict) – Payload for request.
kwargs (dict) – Additional arguments

Returns:

Response the request.

Return type:

Response

Raises:

KeycloakConnectionError – HttpError Can’t connect to server.

raw_put(path, data, **kwargs)[source]¶

Submit put request to the path.

Parameters:

path (str) – Path for request.
data (dict) – Payload for request.
kwargs (dict) – Additional arguments

Returns:

Response the request.

Return type:

Response

Raises:

KeycloakConnectionError – HttpError Can’t connect to server.

raw_delete(path, data=None, **kwargs)[source]¶

Submit delete request to the path.

Parameters:

path (str) – Path for request.
data (dict | None) – Payload for request.
kwargs (dict) – Additional arguments

Returns:

Response the request.

Return type:

Response

Raises:

KeycloakConnectionError – HttpError Can’t connect to server.

async a_raw_get(path, **kwargs)[source]¶

Submit get request to the path.

Parameters:

path (str) – Path for request.
kwargs (dict) – Additional arguments

Returns:

Response the request.

Return type:

Response

Raises:

KeycloakConnectionError – HttpError Can’t connect to server.

async a_raw_post(path, data, **kwargs)[source]¶

Submit post request to the path.

Parameters:

path (str) – Path for request.
data (dict) – Payload for request.
kwargs (dict) – Additional arguments

Returns:

Response the request.

Return type:

Response

Raises:

KeycloakConnectionError – HttpError Can’t connect to server.

async a_raw_put(path, data, **kwargs)[source]¶

Submit put request to the path.

Parameters:

path (str) – Path for request.
data (dict) – Payload for request.
kwargs (dict) – Additional arguments

Returns:

Response the request.

Return type:

Response

Raises:

KeycloakConnectionError – HttpError Can’t connect to server.

async a_raw_delete(path, data=None, **kwargs)[source]¶

Submit delete request to the path.

Parameters:

path (str) – Path for request.
data (dict | None) – Payload for request.
kwargs (dict) – Additional arguments

Returns:

Response the request.

Return type:

Response

Raises:

KeycloakConnectionError – HttpError Can’t connect to server.

exception keycloak.keycloak_uma.KeycloakDeleteError(error_message='', response_code=None, response_body=None)[source]¶

Bases: KeycloakOperationError

Keycloak request delete error exception.

exception keycloak.keycloak_uma.KeycloakGetError(error_message='', response_code=None, response_body=None)[source]¶

Bases: KeycloakOperationError

Keycloak request get error exception.

exception keycloak.keycloak_uma.KeycloakPostError(error_message='', response_code=None, response_body=None)[source]¶

Bases: KeycloakOperationError

Keycloak request post error exception.

exception keycloak.keycloak_uma.KeycloakPutError(error_message='', response_code=None, response_body=None)[source]¶

Bases: KeycloakOperationError

Keycloak request put error exception.

keycloak.keycloak_uma.raise_error_from_response(response, error, expected_codes=None, skip_exists=False)[source]¶

Raise an exception for the response.

Parameters:

response (Response) – The response object
error (dict or Exception) – Error object to raise
expected_codes (Sequence[int]) – Set of expected codes, which should not raise the exception
skip_exists (bool) – Indicates whether the response on already existing object should be ignored

Returns:

Content of the response message

Type:

bytes or dict

Raises:

KeycloakError – In case of unexpected status codes

class keycloak.keycloak_uma.KeycloakOpenIDConnection(server_url, username=None, password=None, token=None, totp=None, realm_name='master', client_id='admin-cli', verify=True, client_secret_key=None, custom_headers=None, user_realm_name=None, timeout=60)[source]¶

Bases: keycloak.connection.ConnectionManager

A class to help with OpenID connections which can auto refresh tokens.

Parameters:: object (_type_) – _description_

_server_url = None¶

_username = None¶

_password = None¶

_totp = None¶

_realm_name = None¶

_client_id = None¶

_verify = None¶

_client_secret_key = None¶

_connection = None¶

_custom_headers = None¶

_user_realm_name = None¶

_expires_at = None¶

_keycloak_openid = None¶

property server_url¶

Get server url.

Returns:: Keycloak server url
Return type:: str

property realm_name¶

Get realm name.

Returns:: Realm name
Return type:: str

property client_id¶

Get client id.

Returns:: Client id
Return type:: str

property client_secret_key¶

Get client secret key.

Returns:: Client secret key
Return type:: str

property username¶

Get username.

Returns:: Admin username
Return type:: str

property password¶

Get password.

Returns:: Admin password
Return type:: str

property totp¶

Get totp.

Returns:: TOTP
Return type:: str

property token¶

Get token.

Returns:: Access and refresh token
Return type:: dict

property expires_at¶

Get token expiry time.

Returns:: Datetime at which the current token will expire
Return type:: datetime

property user_realm_name¶

Get user realm name.

Returns:: User realm name
Return type:: str

property custom_headers¶

Get custom headers.

Returns:: Custom headers
Return type:: dict

property keycloak_openid: keycloak.keycloak_openid.KeycloakOpenID¶

Get the KeycloakOpenID object.

The KeycloakOpenID is used to refresh tokens

Returns:: KeycloakOpenID
Return type:: KeycloakOpenID

get_token()[source]¶

Get admin token.

The admin token is then set in the token attribute.

refresh_token()[source]¶

Refresh the token.

Raises:: KeycloakPostError – In case the refresh token request failed.

_refresh_if_required()[source]¶

raw_get(*args, **kwargs)[source]¶

Call connection.raw_get.

If auto_refresh is set for get and access_token is expired, it will refresh the token and try get once more.

Parameters:

args (tuple) – Additional arguments
kwargs (dict) – Additional keyword arguments

Returns:

Response

Return type:

Response

raw_post(*args, **kwargs)[source]¶

Call connection.raw_post.

If auto_refresh is set for post and access_token is expired, it will refresh the token and try post once more.

Parameters:

args (tuple) – Additional arguments
kwargs (dict) – Additional keyword arguments

Returns:

Response

Return type:

Response

raw_put(*args, **kwargs)[source]¶

Call connection.raw_put.

If auto_refresh is set for put and access_token is expired, it will refresh the token and try put once more.

Parameters:

args (tuple) – Additional arguments
kwargs (dict) – Additional keyword arguments

Returns:

Response

Return type:

Response

raw_delete(*args, **kwargs)[source]¶

Call connection.raw_delete.

If auto_refresh is set for delete and access_token is expired, it will refresh the token and try delete once more.

Parameters:

args (tuple) – Additional arguments
kwargs (dict) – Additional keyword arguments

Returns:

Response

Return type:

Response

async a_get_token()[source]¶

Get admin token.

The admin token is then set in the token attribute.

async a_refresh_token()[source]¶

Refresh the token.

Raises:: KeycloakPostError – In case the refresh token request failed.

async a__refresh_if_required()[source]¶: Refresh the token if it is expired.

async a_raw_get(*args, **kwargs)[source]¶

Call connection.raw_get.

If auto_refresh is set for get and access_token is expired, it will refresh the token and try get once more.

Parameters:

args (tuple) – Additional arguments
kwargs (dict) – Additional keyword arguments

Returns:

Response

Return type:

Response

async a_raw_post(*args, **kwargs)[source]¶

Call connection.raw_post.

If auto_refresh is set for post and access_token is expired, it will refresh the token and try post once more.

Parameters:

args (tuple) – Additional arguments
kwargs (dict) – Additional keyword arguments

Returns:

Response

Return type:

Response

async a_raw_put(*args, **kwargs)[source]¶

Call connection.raw_put.

If auto_refresh is set for put and access_token is expired, it will refresh the token and try put once more.

Parameters:

args (tuple) – Additional arguments
kwargs (dict) – Additional keyword arguments

Returns:

Response

Return type:

Response

async a_raw_delete(*args, **kwargs)[source]¶

Call connection.raw_delete.

If auto_refresh is set for delete and access_token is expired, it will refresh the token and try delete once more.

Parameters:

args (tuple) – Additional arguments
kwargs (dict) – Additional keyword arguments

Returns:

Response

Return type:

Response

class keycloak.keycloak_uma.UMAPermission(permission=None, resource='', scope='')[source]¶

A class to conveniently assemble permissions.

The class itself is callable, and will return the assembled permission.

Usage example:

>>> r = Resource("Users")
>>> s = Scope("delete")
>>> permission = r(s)
>>> print(permission)
    'Users#delete'

Parameters:

permission (UMAPermission) – Permission
resource (str) – Resource
scope (str) – Scope

__str__()[source]¶

Str method.

Returns:: String representation
Return type:: str

__eq__(__o: object) → bool[source]¶

Eq method.

Parameters:: __o (object) – The other object
Returns:: Equality boolean
Return type:: bool

__repr__() → str[source]¶

Repr method.

Returns:: The object representation
Return type:: str

__hash__() → int[source]¶

Hash method.

Returns:: Hash of the object
Return type:: int

__call__(permission=None, resource='', scope='') → UMAPermission[source]¶

Call method.

Parameters:

permission (UMAPermission) – Permission
resource (str) – Resource
scope (str) – Scope

Returns:

The combined UMA permission

Return type:

UMAPermission

Raises:

PermissionDefinitionError – In case bad permission definition

keycloak.keycloak_uma.URL_UMA_WELL_KNOWN[source]¶

class keycloak.keycloak_uma.KeycloakUMA(connection: keycloak.openid_connection.KeycloakOpenIDConnection)[source]¶

Keycloak UMA client.

Parameters:: connection – OpenID connection manager

_fetch_well_known()[source]¶

static format_url(url, **kwargs)[source]¶

Substitute url path parameters.

Given a parameterized url string, returns the string after url encoding and substituting the given params. For example, format_url(“https://myserver/{my_resource}/{id}”, my_resource=”hello world”, id=”myid”) would produce https://myserver/hello+world/myid.

Parameters:

url (str) – url string to format
kwargs (dict) – dict containing kwargs to substitute

Returns:

formatted string

Return type:

str

static a_format_url(url, **kwargs)[source]¶

Async:

Substitute url path parameters.

Given a parameterized url string, returns the string after url encoding and substituting the given params. For example, format_url(“https://myserver/{my_resource}/{id}”, my_resource=”hello world”, id=”myid”) would produce https://myserver/hello+world/myid.

Parameters:

url (str) – url string to format
kwargs (dict) – dict containing kwargs to substitute

Returns:

formatted string

Return type:

str

property uma_well_known[source]¶

Get the well_known UMA2 config.

Returns:: It lists endpoints and other configuration options relevant
Return type:: dict

async a_uma_well_known()[source]¶

Get the well_known UMA2 config async.

Returns:: It lists endpoints and other configuration options relevant
Return type:: dict

resource_set_create(payload)[source]¶

Create a resource set.

Spec https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#rfc.section.2.2.1

ResourceRepresentation https://www.keycloak.org/docs-api/24.0.2/rest-api/index.html#_resourcerepresentation

Parameters:: payload (dict) – ResourceRepresentation
Returns:: ResourceRepresentation with the _id property assigned
Return type:: dict

resource_set_update(resource_id, payload)[source]¶

Update a resource set.

Spec https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#update-resource-set

ResourceRepresentation https://www.keycloak.org/docs-api/24.0.2/rest-api/index.html#_resourcerepresentation

Parameters:

resource_id (str) – id of the resource
payload (dict) – ResourceRepresentation

Returns:

Response dict (empty)

Return type:

dict

resource_set_read(resource_id)[source]¶

Read a resource set.

Spec https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#read-resource-set

ResourceRepresentation https://www.keycloak.org/docs-api/24.0.2/rest-api/index.html#_resourcerepresentation

Parameters:: resource_id (str) – id of the resource
Returns:: ResourceRepresentation
Return type:: dict

resource_set_delete(resource_id)[source]¶

Delete a resource set.

Spec https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#delete-resource-set

Parameters:: resource_id (str) – id of the resource
Returns:: Response dict (empty)
Return type:: dict

resource_set_list_ids(name: str = '', exact_name: bool = False, uri: str = '', owner: str = '', resource_type: str = '', scope: str = '', first: int = 0, maximum: int = -1)[source]¶

Query for list of resource set ids.

Spec https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#list-resource-sets

Parameters:

name (str) – query resource name
exact_name (bool) – query exact match for resource name
uri (str) – query resource uri
owner (str) – query resource owner
resource_type (str) – query resource type
scope (str) – query resource scope
first (int) – index of first matching resource to return
maximum (int) – maximum number of resources to return (-1 for all)

Returns:

List of ids

Return type:

List[str]

resource_set_list()[source]¶

List all resource sets.

Spec https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#list-resource-sets

ResourceRepresentation https://www.keycloak.org/docs-api/24.0.2/rest-api/index.html#_resourcerepresentation

Yields:: Iterator over a list of ResourceRepresentations
Return type:: Iterator[dict]

permission_ticket_create(permissions: Iterable[keycloak.uma_permissions.UMAPermission])[source]¶

Create a permission ticket.

Parameters:: permissions (Iterable[UMAPermission]) – Iterable of uma permissions to validate the token against
Returns:: Keycloak decision
Return type:: boolean
Raises:: KeycloakPostError – In case permission resource not found

permissions_check(token, permissions: Iterable[keycloak.uma_permissions.UMAPermission])[source]¶

Check UMA permissions by user token with requested permissions.

The token endpoint is used to check UMA permissions from Keycloak. It can only be invoked by confidential clients.

https://www.keycloak.org/docs/latest/authorization_services/#_service_authorization_api

Parameters:

token (str) – user token
permissions (Iterable[UMAPermission]) – Iterable of uma permissions to validate the token against

Returns:

Keycloak decision

Return type:

boolean

policy_resource_create(resource_id, payload)[source]¶

Create permission policy for resource.

Supports name, description, scopes, roles, groups, clients

https://www.keycloak.org/docs/latest/authorization_services/#associating-a-permission-with-a-resource

Parameters:

resource_id (str) – _id of resource
payload (dict) – permission configuration

Returns:

PermissionRepresentation

Return type:

dict

policy_update(policy_id, payload)[source]¶

Update permission policy.

https://www.keycloak.org/docs/latest/authorization_services/#associating-a-permission-with-a-resource https://www.keycloak.org/docs-api/24.0.2/rest-api/index.html#_policyrepresentation

Parameters:

policy_id (str) – id of policy permission
payload (dict) – policy permission configuration

Returns:

PermissionRepresentation

Return type:

dict

policy_delete(policy_id)[source]¶

Delete permission policy.

https://www.keycloak.org/docs/latest/authorization_services/#removing-a-permission https://www.keycloak.org/docs-api/24.0.2/rest-api/index.html#_policyrepresentation

Parameters:: policy_id (str) – id of permission policy
Returns:: PermissionRepresentation
Return type:: dict

policy_query(resource: str = '', name: str = '', scope: str = '', first: int = 0, maximum: int = -1)[source]¶

Query permission policies.

https://www.keycloak.org/docs/latest/authorization_services/#querying-permission

Parameters:

resource (str) – query resource id
name (str) – query resource name
scope (str) – query resource scope
first (int) – index of first matching resource to return
maximum (int) – maximum number of resources to return (-1 for all)

Returns:

List of ids

Returns:

List of ids

Return type:

List[str]

async a__fetch_well_known()[source]¶

Get the well_known UMA2 config async.

Returns:: It lists endpoints and other configuration options relevant
Return type:: dict

async a_resource_set_create(payload)[source]¶

Create a resource set asynchronously.

Spec https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#rfc.section.2.2.1

ResourceRepresentation https://www.keycloak.org/docs-api/24.0.2/rest-api/index.html#_resourcerepresentation

Parameters:: payload (dict) – ResourceRepresentation
Returns:: ResourceRepresentation with the _id property assigned
Return type:: dict

async a_resource_set_update(resource_id, payload)[source]¶

Update a resource set asynchronously.

Spec https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#update-resource-set

ResourceRepresentation https://www.keycloak.org/docs-api/24.0.2/rest-api/index.html#_resourcerepresentation

Parameters:

resource_id (str) – id of the resource
payload (dict) – ResourceRepresentation

Returns:

Response dict (empty)

Return type:

dict

async a_resource_set_read(resource_id)[source]¶

Read a resource set asynchronously.

Spec https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#read-resource-set

ResourceRepresentation https://www.keycloak.org/docs-api/24.0.2/rest-api/index.html#_resourcerepresentation

Parameters:: resource_id (str) – id of the resource
Returns:: ResourceRepresentation
Return type:: dict

async a_resource_set_delete(resource_id)[source]¶

Delete a resource set asynchronously.

Spec https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#delete-resource-set

Parameters:: resource_id (str) – id of the resource
Returns:: Response dict (empty)
Return type:: dict

async a_resource_set_list_ids(name: str = '', exact_name: bool = False, uri: str = '', owner: str = '', resource_type: str = '', scope: str = '', first: int = 0, maximum: int = -1)[source]¶

Query for list of resource set ids asynchronously.

Spec https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#list-resource-sets

Parameters:

name (str) – query resource name
exact_name (bool) – query exact match for resource name
uri (str) – query resource uri
owner (str) – query resource owner
resource_type (str) – query resource type
scope (str) – query resource scope
first (int) – index of first matching resource to return
maximum (int) – maximum number of resources to return (-1 for all)

Returns:

List of ids

Return type:

List[str]

async a_resource_set_list()[source]¶

List all resource sets asynchronously.

Spec https://docs.kantarainitiative.org/uma/rec-oauth-resource-reg-v1_0_1.html#list-resource-sets

ResourceRepresentation https://www.keycloak.org/docs-api/24.0.2/rest-api/index.html#_resourcerepresentation

Yields:: Iterator over a list of ResourceRepresentations
Return type:: Iterator[dict]

async a_permission_ticket_create(permissions: Iterable[keycloak.uma_permissions.UMAPermission])[source]¶

Create a permission ticket asynchronously.

Parameters:: permissions (Iterable[UMAPermission]) – Iterable of uma permissions to validate the token against
Returns:: Keycloak decision
Return type:: boolean
Raises:: KeycloakPostError – In case permission resource not found

async a_permissions_check(token, permissions: Iterable[keycloak.uma_permissions.UMAPermission])[source]¶

Check UMA permissions by user token with requested permissions asynchronously.

The token endpoint is used to check UMA permissions from Keycloak. It can only be invoked by confidential clients.

https://www.keycloak.org/docs/latest/authorization_services/#_service_authorization_api

Parameters:

token (str) – user token
permissions (Iterable[UMAPermission]) – Iterable of uma permissions to validate the token against

Returns:

Keycloak decision

Return type:

boolean

async a_policy_resource_create(resource_id, payload)[source]¶

Create permission policy for resource asynchronously.

Supports name, description, scopes, roles, groups, clients

https://www.keycloak.org/docs/latest/authorization_services/#associating-a-permission-with-a-resource

Parameters:

resource_id (str) – _id of resource
payload (dict) – permission configuration

Returns:

PermissionRepresentation

Return type:

dict

async a_policy_update(policy_id, payload)[source]¶

Update permission policy asynchronously.

https://www.keycloak.org/docs/latest/authorization_services/#associating-a-permission-with-a-resource https://www.keycloak.org/docs-api/24.0.2/rest-api/index.html#_policyrepresentation

Parameters:

policy_id (str) – id of policy permission
payload (dict) – policy permission configuration

Returns:

PermissionRepresentation

Return type:

dict

async a_policy_delete(policy_id)[source]¶

Delete permission policy asynchronously.

https://www.keycloak.org/docs/latest/authorization_services/#removing-a-permission https://www.keycloak.org/docs-api/24.0.2/rest-api/index.html#_policyrepresentation

Parameters:: policy_id (str) – id of permission policy
Returns:: PermissionRepresentation
Return type:: dict

async a_policy_query(resource: str = '', name: str = '', scope: str = '', first: int = 0, maximum: int = -1)[source]¶

Query permission policies asynchronously.

https://www.keycloak.org/docs/latest/authorization_services/#querying-permission

Parameters:

resource (str) – query resource id
name (str) – query resource name
scope (str) – query resource scope
first (int) – index of first matching resource to return
maximum (int) – maximum number of resources to return (-1 for all)

Returns:

List of ids

Returns:

List of ids

Return type:

List[str]