Browse Source

Merge branch 'master' into add-client-session-stats

pull/258/head
Marcos Pereira 3 years ago
committed by GitHub
parent
commit
575dba520e
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
  1. 15
      README.md
  2. 2
      bin/deploy.sh
  3. 12
      docs/source/index.rst
  4. 6
      keycloak/connection.py
  5. 395
      keycloak/keycloak_admin.py
  6. 6
      keycloak/keycloak_openid.py
  7. 18
      keycloak/urls_patterns.py

15
README.md

@ -173,6 +173,15 @@ response = keycloak_admin.update_user(user_id="user-id-keycloak",
# Update User Password
response = keycloak_admin.set_user_password(user_id="user-id-keycloak", password="secret", temporary=True)
# Get User Credentials
credentials = keycloak_admin.get_credentials(user_id='user_id')
# Get User Credential by ID
credential = keycloak_admin.get_credential(user_id='user_id', credential_id='credential_id')
# Delete User Credential
response = keycloak_admin.delete_credential(user_id='user_id', credential_id='credential_id')
# Delete User
response = keycloak_admin.delete_user(user_id="user-id-keycloak")
@ -234,7 +243,7 @@ keycloak_admin.delete_client_roles_of_user(client_id="client_id", user_id="user_
keycloak_admin.delete_client_roles_of_user(client_id="client_id", user_id="user_id", roles=[{"id": "role-id_1"}, {"id": "role-id_2"}])
# Create new group
group = keycloak_admin.create_group(name="Example Group")
group = keycloak_admin.create_group({"name": "Example Group"})
# Get all groups
groups = keycloak_admin.get_groups()
@ -257,6 +266,10 @@ realm_roles = keycloak_admin.get_roles()
# Assign client role to user. Note that BOTH role_name and role_id appear to be required.
keycloak_admin.assign_client_role(client_id=client_id, user_id=user_id, role_id=role_id, role_name="test")
# Assign realm roles to user
keycloak_admin.assign_realm_roles(user_id=user_id, roles=realm_roles)
# Get all ID Providers
idps = keycloak_admin.get_idps()

2
bin/deploy.sh

@ -9,5 +9,5 @@ username=${PYPI_USERNAME}
password=${PYPI_PASSWORD}
EOF
python setup.py sdist
python setup.py sdist bdist_wheel
twine upload dist/*

12
docs/source/index.rst

@ -100,6 +100,15 @@ Main methods::
# verify=True,
# custom_headers={'CustomHeader': 'value'})
# Optionally, you can pass proxies as well that will be used in all HTTP calls. See requests documentation for more details_
# `requests-proxies <https://2.python-requests.org/en/master/user/advanced/#id10>`_.
# keycloak_openid = KeycloakOpenID(server_url="http://localhost:8080/auth/",
# client_id="example_client",
# realm_name="example_realm",
# client_secret_key="secret",
# verify=True,
# proxies={'http': 'http://10.10.1.10:3128', 'https': 'http://10.10.1.10:1080'})
# Get WellKnow
config_well_know = keycloak_openid.well_know()
@ -262,6 +271,9 @@ Main methods::
# Assign realm roles to user. Note that BOTH role_name and role_id appear to be required.
keycloak_admin.assign_realm_roles(client_id="client_id", user_id="user_id", roles=[{"roles_representation"}])
# Delete realm roles of user. Note that BOTH role_name and role_id appear to be required.
keycloak_admin.deletes_realm_roles_of_user(user_id="user_id", roles=[{"roles_representation"}])
# Create new group
group = keycloak_admin.create_group(name="Example Group")

6
keycloak/connection.py

@ -39,9 +39,10 @@ class ConnectionManager(object):
headers (dict): The header parameters of the requests to the server.
timeout (int): Timeout to use for requests to the server.
verify (bool): Verify server SSL.
proxies (dict): The proxies servers requests is sent by.
"""
def __init__(self, base_url, headers={}, timeout=60, verify=True):
def __init__(self, base_url, headers={}, timeout=60, verify=True, proxies=None):
self._base_url = base_url
self._headers = headers
self._timeout = timeout
@ -60,6 +61,9 @@ class ConnectionManager(object):
self._s.mount(protocol, adapter)
if proxies:
self._s.proxies.update(proxies)
def __del__(self):
self._s.close()

395
keycloak/keycloak_admin.py

@ -32,6 +32,7 @@ from .connection import ConnectionManager
from .exceptions import raise_error_from_response, KeycloakGetError
from .keycloak_openid import KeycloakOpenID
from .urls_patterns import URL_ADMIN_SERVER_INFO, URL_ADMIN_CLIENT_AUTHZ_RESOURCES, URL_ADMIN_CLIENT_ROLES, \
URL_ADMIN_CLIENT_AUTHZ_POLICIES, URL_ADMIN_CLIENT_AUTHZ_ROLE_BASED_POLICY, URL_ADMIN_CLIENT_AUTHZ_RESOURCE_BASED_PERMISSION, \
URL_ADMIN_GET_SESSIONS, URL_ADMIN_RESET_PASSWORD, URL_ADMIN_SEND_UPDATE_ACCOUNT, URL_ADMIN_GROUPS_REALM_ROLES, \
URL_ADMIN_REALM_ROLES_COMPOSITE_REALM_ROLE, URL_ADMIN_CLIENT_INSTALLATION_PROVIDER, \
URL_ADMIN_REALM_ROLES_ROLE_BY_NAME, URL_ADMIN_GROUPS_CLIENT_ROLES, \
@ -45,10 +46,13 @@ from .urls_patterns import URL_ADMIN_SERVER_INFO, URL_ADMIN_CLIENT_AUTHZ_RESOURC
URL_ADMIN_USER_REALM_ROLES, URL_ADMIN_REALM, URL_ADMIN_COMPONENTS, URL_ADMIN_COMPONENT, URL_ADMIN_KEYS, \
URL_ADMIN_USER_FEDERATED_IDENTITY, URL_ADMIN_USER_FEDERATED_IDENTITIES, URL_ADMIN_CLIENT_ROLE_MEMBERS, \
URL_ADMIN_REALM_ROLES_MEMBERS, URL_ADMIN_CLIENT_PROTOCOL_MAPPER, URL_ADMIN_CLIENT_SCOPES_MAPPERS, \
URL_ADMIN_FLOWS_EXECUTIONS_EXEUCUTION, URL_ADMIN_FLOWS_EXECUTIONS_FLOW, URL_ADMIN_FLOWS_COPY, \
URL_ADMIN_FLOWS_EXECUTIONS_EXECUTION, URL_ADMIN_FLOWS_EXECUTIONS_FLOW, URL_ADMIN_FLOWS_COPY, \
URL_ADMIN_FLOWS_ALIAS, URL_ADMIN_CLIENT_SERVICE_ACCOUNT_USER, URL_ADMIN_AUTHENTICATOR_CONFIG, \
URL_ADMIN_CLIENT_ROLES_COMPOSITE_CLIENT_ROLE, URL_ADMIN_CLIENT_ALL_SESSIONS, URL_ADMIN_EVENTS, \
URL_ADMIN_REALM_EXPORT, URL_ADMIN_DELETE_USER_ROLE, URL_ADMIN_USER_LOGOUT
URL_ADMIN_REALM_EXPORT, URL_ADMIN_DELETE_USER_ROLE, URL_ADMIN_USER_LOGOUT, URL_ADMIN_FLOWS_EXECUTION, \
URL_ADMIN_FLOW, URL_ADMIN_DEFAULT_DEFAULT_CLIENT_SCOPES, URL_ADMIN_DEFAULT_DEFAULT_CLIENT_SCOPE, \
URL_ADMIN_DEFAULT_OPTIONAL_CLIENT_SCOPES, URL_ADMIN_DEFAULT_OPTIONAL_CLIENT_SCOPE, \
URL_ADMIN_USER_CREDENTIALS, URL_ADMIN_USER_CREDENTIAL
class KeycloakAdmin:
@ -229,6 +233,13 @@ class KeycloakAdmin:
page += 1
return results
def __fetch_paginated(self, url, query=None):
query = query or {}
return raise_error_from_response(
self.raw_get(url, **query),
KeycloakGetError)
def import_realm(self, payload):
"""
Import a new realm from a RealmRepresentation. Realm name must be unique.
@ -326,8 +337,14 @@ class KeycloakAdmin:
:param query: Query parameters (optional)
:return: users list
"""
query = query or {}
params_path = {"realm-name": self.realm_name}
return self.__fetch_all(URL_ADMIN_USERS.format(**params_path), query)
url = URL_ADMIN_USERS.format(**params_path)
if "first" in query or "max" in query:
return self.__fetch_paginated(url, query)
return self.__fetch_all(url, query)
def create_idp(self, payload):
"""
@ -506,6 +523,50 @@ class KeycloakAdmin:
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def get_credentials(self, user_id):
"""
Returns a list of credential belonging to the user.
CredentialRepresentation
https://www.keycloak.org/docs-api/8.0/rest-api/index.html#_credentialrepresentation
:param: user_id: user id
:return: Keycloak server response (CredentialRepresentation)
"""
params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.raw_get(URL_ADMIN_USER_CREDENTIALS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def get_credential(self, user_id, credential_id):
"""
Get credential of the user.
CredentialRepresentation
https://www.keycloak.org/docs-api/8.0/rest-api/index.html#_credentialrepresentation
:param: user_id: user id
:param: credential_id: credential id
:return: Keycloak server response (ClientRepresentation)
"""
params_path = {"realm-name": self.realm_name, "id": user_id, "credential_id": credential_id}
data_raw = self.raw_get(URL_ADMIN_USER_CREDENTIAL.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def delete_credential(self, user_id, credential_id):
"""
Delete credential of the user.
CredentialRepresentation
https://www.keycloak.org/docs-api/8.0/rest-api/index.html#_credentialrepresentation
:param: user_id: user id
:param: credential_id: credential id
:return: Keycloak server response (ClientRepresentation)
"""
params_path = {"realm-name": self.realm_name, "id": user_id, "credential_id": credential_id}
data_raw = self.raw_delete(URL_ADMIN_USER_CREDENTIAL.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def logout(self, user_id):
"""
Logs out user.
@ -555,6 +616,18 @@ class KeycloakAdmin:
params_path = {"realm-name": self.realm_name, "id": user_id, "provider": provider_id}
data_raw = self.raw_post(URL_ADMIN_USER_FEDERATED_IDENTITY.format(**params_path), data=json.dumps(payload))
def delete_user_social_login(self, user_id, provider_id):
"""
Delete a federated identity / social login provider from the user
:param user_id: User id
:param provider_id: Social login provider id
:return:
"""
params_path = {"realm-name": self.realm_name, "id": user_id, "provider": provider_id}
data_raw = self.raw_delete(URL_ADMIN_USER_FEDERATED_IDENTITY.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def send_update_account(self, user_id, payload, client_id=None, lifespan=None, redirect_uri=None):
"""
Send an update account email to the user. An email contains a
@ -618,7 +691,7 @@ class KeycloakAdmin:
data_raw = self.raw_get(URL_ADMIN_SERVER_INFO)
return raise_error_from_response(data_raw, KeycloakGetError)
def get_groups(self):
def get_groups(self, query=None):
"""
Returns a list of groups belonging to the realm
@ -627,8 +700,14 @@ class KeycloakAdmin:
:return: array GroupRepresentation
"""
query = query or {}
params_path = {"realm-name": self.realm_name}
return self.__fetch_all(URL_ADMIN_GROUPS.format(**params_path))
url = URL_ADMIN_USERS.format(**params_path)
if "first" in query or "max" in query:
return self.__fetch_paginated(url, query)
return self.__fetch_all(url, query)
def get_group(self, group_id):
"""
@ -680,7 +759,12 @@ class KeycloakAdmin:
:return: Keycloak server response (UserRepresentation)
"""
params_path = {"realm-name": self.realm_name, "id": group_id}
return self.__fetch_all(URL_ADMIN_GROUP_MEMBERS.format(**params_path), query)
url = URL_ADMIN_USERS.format(**params_path)
if "first" in query or "max" in query:
return self.__fetch_paginated(url, query)
return self.__fetch_all(url, query)
def get_group_by_path(self, path, search_in_subgroups=False):
"""
@ -866,6 +950,25 @@ class KeycloakAdmin:
data_raw = self.raw_get(URL_ADMIN_CLIENT_AUTHZ_SETTINGS.format(**params_path))
return data_raw
def create_client_authz_resource(self, client_id, payload, skip_exists=False):
"""
Create resources of client.
:param client_id: id in ClientRepresentation
https://www.keycloak.org/docs-api/8.0/rest-api/index.html#_clientrepresentation
:param payload: ResourceRepresentation
https://www.keycloak.org/docs-api/12.0/rest-api/index.html#_resourcerepresentation
:return: Keycloak server response
"""
params_path = {"realm-name": self.realm_name,
"id": client_id}
data_raw = self.raw_post(URL_ADMIN_CLIENT_AUTHZ_RESOURCES.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[201], skip_exists=skip_exists)
def get_client_authz_resources(self, client_id):
"""
Get resources from client.
@ -877,7 +980,85 @@ class KeycloakAdmin:
params_path = {"realm-name": self.realm_name, "id": client_id}
data_raw = self.raw_get(URL_ADMIN_CLIENT_AUTHZ_RESOURCES.format(**params_path))
return data_raw
return raise_error_from_response(data_raw, KeycloakGetError)
def create_client_authz_role_based_policy(self, client_id, payload, skip_exists=False):
"""
Create role-based policy of client.
:param client_id: id in ClientRepresentation
https://www.keycloak.org/docs-api/8.0/rest-api/index.html#_clientrepresentation
:param payload: No Document
payload example:
payload={
"type": "role",
"logic": "POSITIVE",
"decisionStrategy": "UNANIMOUS",
"name": "Policy-1",
"roles": [
{
"id": id
}
]
}
:return: Keycloak server response
"""
params_path = {"realm-name": self.realm_name,
"id": client_id}
data_raw = self.raw_post(URL_ADMIN_CLIENT_AUTHZ_ROLE_BASED_POLICY.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[201], skip_exists=skip_exists)
def create_client_authz_resource_based_permission(self, client_id, payload, skip_exists=False):
"""
Create resource-based permission of client.
:param client_id: id in ClientRepresentation
https://www.keycloak.org/docs-api/8.0/rest-api/index.html#_clientrepresentation
:param payload: PolicyRepresentation
https://www.keycloak.org/docs-api/12.0/rest-api/index.html#_policyrepresentation
payload example:
payload={
"type": "resource",
"logic": "POSITIVE",
"decisionStrategy": "UNANIMOUS",
"name": "Permission-Name",
"resources": [
resource_id
],
"policies": [
policy_id
]
:return: Keycloak server response
"""
params_path = {"realm-name": self.realm_name,
"id": client_id}
data_raw = self.raw_post(URL_ADMIN_CLIENT_AUTHZ_RESOURCE_BASED_PERMISSION.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[201], skip_exists=skip_exists)
def get_client_authz_policies(self, client_id):
"""
Get policies from client.
:param client_id: id in ClientRepresentation
https://www.keycloak.org/docs-api/8.0/rest-api/index.html#_clientrepresentation
:param payload: PolicyRepresentation
https://www.keycloak.org/docs-api/12.0/rest-api/index.html#_policyrepresentation
:return: Keycloak server response
"""
params_path = {"realm-name": self.realm_name, "id": client_id}
params_query = {"first": 0, "max": 20, "permission": False}
data_raw = self.raw_get(URL_ADMIN_CLIENT_AUTHZ_POLICIES.format(**params_path), **params_query)
return raise_error_from_response(data_raw, KeycloakGetError)
def get_client_service_account_user(self, client_id):
"""
@ -1210,7 +1391,6 @@ class KeycloakAdmin:
Assign realm roles to a user
:param user_id: id of user
:param client_id: id of client containing role (not client-id)
:param roles: roles list or role (use RoleRepresentation)
:return Keycloak server response
"""
@ -1221,6 +1401,21 @@ class KeycloakAdmin:
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def delete_realm_roles_of_user(self, user_id, roles):
"""
Deletes realm roles of a user
:param user_id: id of user
:param roles: roles list or role (use RoleRepresentation)
:return Keycloak server response
"""
payload = roles if isinstance(roles, list) else [roles]
params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.raw_delete(URL_ADMIN_USER_REALM_ROLES.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def get_realm_roles_of_user(self, user_id):
"""
Get all realm roles for a user.
@ -1427,6 +1622,20 @@ class KeycloakAdmin:
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[201])
def delete_authentication_flow(self, flow_id):
"""
Delete authentication flow
AuthenticationInfoRepresentation
https://www.keycloak.org/docs-api/8.0/rest-api/index.html#_authenticationinforepresentation
:param flow_id: authentication flow id
:return: Keycloak server response
"""
params_path = {"realm-name": self.realm_name, "id": flow_id}
data_raw = self.raw_delete(URL_ADMIN_FLOW.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def get_authentication_flow_executions(self, flow_alias):
"""
Get authentication flow executions. Returns all execution steps
@ -1455,6 +1664,20 @@ class KeycloakAdmin:
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def get_authentication_flow_execution(self, execution_id):
"""
Get authentication flow execution.
AuthenticationExecutionInfoRepresentation
https://www.keycloak.org/docs-api/8.0/rest-api/index.html#_authenticationexecutioninforepresentation
:param execution_id: the execution ID
:return: Response(json)
"""
params_path = {"realm-name": self.realm_name, "id": execution_id}
data_raw = self.raw_get(URL_ADMIN_FLOWS_EXECUTION.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def create_authentication_flow_execution(self, payload, flow_alias):
"""
Create an authentication flow execution
@ -1468,10 +1691,24 @@ class KeycloakAdmin:
"""
params_path = {"realm-name": self.realm_name, "flow-alias": flow_alias}
data_raw = self.raw_post(URL_ADMIN_FLOWS_EXECUTIONS_EXEUCUTION.format(**params_path),
data_raw = self.raw_post(URL_ADMIN_FLOWS_EXECUTIONS_EXECUTION.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[201])
def delete_authentication_flow_execution(self, execution_id):
"""
Delete authentication flow execution
AuthenticationExecutionInfoRepresentation
https://www.keycloak.org/docs-api/8.0/rest-api/index.html#_authenticationexecutioninforepresentation
:param execution_id: keycloak client id (not oauth client-id)
:return: Keycloak server response (json)
"""
params_path = {"realm-name": self.realm_name, "id": execution_id}
data_raw = self.raw_delete(URL_ADMIN_FLOWS_EXECUTION.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def create_authentication_flow_subflow(self, payload, flow_alias, skip_exists=False):
"""
Create a new sub authentication flow for a given authentication flow
@ -1659,6 +1896,78 @@ class KeycloakAdmin:
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def get_default_default_client_scopes(self):
"""
Return list of default default client scopes
:return: Keycloak server response
"""
params_path = {"realm-name": self.realm_name}
data_raw = self.raw_get(URL_ADMIN_DEFAULT_DEFAULT_CLIENT_SCOPES.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def delete_default_default_client_scope(self, scope_id):
"""
Delete default default client scope
:param scope_id: default default client scope id
:return: Keycloak server response
"""
params_path = {"realm-name": self.realm_name, "id": scope_id}
data_raw = self.raw_delete(URL_ADMIN_DEFAULT_DEFAULT_CLIENT_SCOPE.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def add_default_default_client_scope(self, scope_id):
"""
Add default default client scope
:param scope_id: default default client scope id
:return: Keycloak server response
"""
params_path = {"realm-name": self.realm_name, "id": scope_id}
payload = {"realm": self.realm_name, "clientScopeId": scope_id}
data_raw = self.raw_put(URL_ADMIN_DEFAULT_DEFAULT_CLIENT_SCOPE.format(**params_path), data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def get_default_optional_client_scopes(self):
"""
Return list of default optional client scopes
:return: Keycloak server response
"""
params_path = {"realm-name": self.realm_name}
data_raw = self.raw_get(URL_ADMIN_DEFAULT_OPTIONAL_CLIENT_SCOPES.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def delete_default_optional_client_scope(self, scope_id):
"""
Delete default optional client scope
:param scope_id: default optional client scope id
:return: Keycloak server response
"""
params_path = {"realm-name": self.realm_name, "id": scope_id}
data_raw = self.raw_delete(URL_ADMIN_DEFAULT_OPTIONAL_CLIENT_SCOPE.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def add_default_optional_client_scope(self, scope_id):
"""
Add default optional client scope
:param scope_id: default optional client scope id
:return: Keycloak server response
"""
params_path = {"realm-name": self.realm_name, "id": scope_id}
payload = {"realm": self.realm_name, "clientScopeId": scope_id}
data_raw = self.raw_put(URL_ADMIN_DEFAULT_OPTIONAL_CLIENT_SCOPE.format(**params_path), data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def add_mapper_to_client(self, client_id, payload):
"""
Add a mapper to a client
@ -1672,10 +1981,50 @@ class KeycloakAdmin:
params_path = {"realm-name": self.realm_name, "id": client_id}
data_raw = self.raw_post(
URL_ADMIN_CLIENT_PROTOCOL_MAPPER.format(**params_path), data=json.dumps(payload))
URL_ADMIN_CLIENT_PROTOCOL_MAPPERS.format(**params_path), data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[201])
def update_client_mapper(self, client_id, mapper_id, payload):
"""
Update client mapper
:param client_id: The id of the client
:param client_mapper_id: The id of the mapper to be deleted
:param payload: ProtocolMapperRepresentation
:return: Keycloak server response
"""
params_path = {
"realm-name": self.realm_name,
"id": self.client_id,
"protocol-mapper-id": mapper_id,
}
data_raw = self.raw_put(
URL_ADMIN_CLIENT_PROTOCOL_MAPPER.format(**params_path), data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def remove_client_mapper(self, client_id, client_mapper_id):
"""
Removes a mapper from the client
https://www.keycloak.org/docs-api/15.0/rest-api/index.html#_protocol_mappers_resource
:param client_id: The id of the client
:param client_mapper_id: The id of the mapper to be deleted
:return: Keycloak server response
"""
params_path = {
"realm-name": self.realm_name,
"id": client_id,
"protocol-mapper-id": mapper_id
}
data_raw = self.raw_delete(
URL_ADMIN_CLIENT_PROTOCOL_MAPPER.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError, expected_codes=[204])
def generate_client_secrets(self, client_id):
"""
@ -1873,7 +2222,13 @@ class KeycloakAdmin:
return r
def get_token(self):
token_realm_name = 'master' if self.client_secret_key else self.user_realm_name or self.realm_name
if self.user_realm_name:
token_realm_name = self.user_realm_name
elif self.realm_name:
token_realm_name = self.realm_name
else:
token_realm_name = "master"
self.keycloak_openid = KeycloakOpenID(server_url=self.server_url, client_id=self.client_id,
realm_name=token_realm_name, verify=self.verify,
client_secret_key=self.client_secret_key,
@ -1885,12 +2240,16 @@ class KeycloakAdmin:
if self.user_realm_name:
self.realm_name = self.user_realm_name
if self.username and self.password:
self._token = self.keycloak_openid.token(self.username, self.password, grant_type=grant_type)
headers = {
'Authorization': 'Bearer ' + self.token.get('access_token'),
'Content-Type': 'application/json'
}
else:
self._token = None
headers = {}
if self.custom_headers is not None:
# merge custom headers to main headers
@ -1902,15 +2261,23 @@ class KeycloakAdmin:
verify=self.verify)
def refresh_token(self):
refresh_token = self.token.get('refresh_token')
refresh_token = self.token.get('refresh_token', None)
if refresh_token is None:
self.get_token()
else:
try:
self.token = self.keycloak_openid.refresh_token(refresh_token)
except KeycloakGetError as e:
if e.response_code == 400 and (b'Refresh token expired' in e.response_body or
b'Token is not active' in e.response_body):
list_errors = [
b'Refresh token expired',
b'Token is not active',
b'Session not active'
]
if e.response_code == 400 and any(err in e.response_body for err in list_errors):
self.get_token()
else:
raise
self.connection.add_param_headers('Authorization', 'Bearer ' + self.token.get('access_token'))
def get_client_all_sessions(self, client_id):

6
keycloak/keycloak_openid.py

@ -44,7 +44,7 @@ from .urls_patterns import (
class KeycloakOpenID:
def __init__(self, server_url, realm_name, client_id, client_secret_key=None, verify=True, custom_headers=None):
def __init__(self, server_url, realm_name, client_id, client_secret_key=None, verify=True, custom_headers=None, proxies=None):
"""
:param server_url: Keycloak server url
@ -53,6 +53,7 @@ class KeycloakOpenID:
:param client_secret_key: client secret key
:param verify: True if want check connection SSL
:param custom_headers: dict of custom header to pass to each HTML request
:param proxies: dict of proxies to sent the request by.
"""
self._client_id = client_id
self._client_secret_key = client_secret_key
@ -64,7 +65,8 @@ class KeycloakOpenID:
self._connection = ConnectionManager(base_url=server_url,
headers=headers,
timeout=60,
verify=verify)
verify=verify,
proxies=proxies)
self._authorization = Authorization()

18
keycloak/urls_patterns.py

@ -50,6 +50,8 @@ URL_ADMIN_USER_CLIENT_ROLES_COMPOSITE = "admin/realms/{realm-name}/users/{id}/ro
URL_ADMIN_USER_GROUP = "admin/realms/{realm-name}/users/{id}/groups/{group-id}"
URL_ADMIN_USER_GROUPS = "admin/realms/{realm-name}/users/{id}/groups"
URL_ADMIN_USER_PASSWORD = "admin/realms/{realm-name}/users/{id}/reset-password"
URL_ADMIN_USER_CREDENTIALS = "admin/realms/{realm-name}/users/{id}/credentials"
URL_ADMIN_USER_CREDENTIAL = "admin/realms/{realm-name}/users/{id}/credentials/{credential_id}"
URL_ADMIN_USER_LOGOUT = "admin/realms/{realm-name}/users/{id}/logout"
URL_ADMIN_USER_STORAGE = "admin/realms/{realm-name}/user-storage/{id}/sync"
@ -71,10 +73,15 @@ URL_ADMIN_CLIENT_ROLES_COMPOSITE_CLIENT_ROLE = URL_ADMIN_CLIENT_ROLE + "/composi
URL_ADMIN_CLIENT_ROLE_MEMBERS = URL_ADMIN_CLIENT + "/roles/{role-name}/users"
URL_ADMIN_CLIENT_AUTHZ_SETTINGS = URL_ADMIN_CLIENT + "/authz/resource-server/settings"
URL_ADMIN_CLIENT_AUTHZ_RESOURCES = URL_ADMIN_CLIENT + "/authz/resource-server/resource"
URL_ADMIN_CLIENT_AUTHZ_POLICIES = URL_ADMIN_CLIENT + "/authz/resource-server/policy"
URL_ADMIN_CLIENT_AUTHZ_ROLE_BASED_POLICY = URL_ADMIN_CLIENT_AUTHZ_POLICIES + "/role"
URL_ADMIN_CLIENT_AUTHZ_PERMISSIONS = URL_ADMIN_CLIENT + "/authz/resource-server/permission"
URL_ADMIN_CLIENT_AUTHZ_RESOURCE_BASED_PERMISSION = URL_ADMIN_CLIENT_AUTHZ_PERMISSIONS + "/resource"
URL_ADMIN_CLIENT_SERVICE_ACCOUNT_USER = URL_ADMIN_CLIENT + "/service-account-user"
URL_ADMIN_CLIENT_CERTS = URL_ADMIN_CLIENT + "/certificates/{attr}"
URL_ADMIN_CLIENT_INSTALLATION_PROVIDER = URL_ADMIN_CLIENT + "/installation/providers/{provider-id}"
URL_ADMIN_CLIENT_PROTOCOL_MAPPER = URL_ADMIN_CLIENT + "/protocol-mappers/models"
URL_ADMIN_CLIENT_PROTOCOL_MAPPERS = URL_ADMIN_CLIENT + "/protocol-mappers/models"
URL_ADMIN_CLIENT_PROTOCOL_MAPPER = URL_ADMIN_CLIENT_PROTOCOL_MAPPERS + "/{protocol-mapper-id}"
URL_ADMIN_CLIENT_SCOPES = "admin/realms/{realm-name}/client-scopes"
URL_ADMIN_CLIENT_SCOPE = URL_ADMIN_CLIENT_SCOPES + "/{scope-id}"
@ -92,11 +99,18 @@ URL_ADMIN_REALM_ROLES_ROLE_BY_NAME = "admin/realms/{realm-name}/roles/{role-name
URL_ADMIN_REALM_ROLES_COMPOSITE_REALM_ROLE = "admin/realms/{realm-name}/roles/{role-name}/composites"
URL_ADMIN_REALM_EXPORT = "admin/realms/{realm-name}/partial-export?exportClients={export-clients}&exportGroupsAndRoles={export-groups-and-roles}"
URL_ADMIN_DEFAULT_DEFAULT_CLIENT_SCOPES = URL_ADMIN_REALM + "/default-default-client-scopes"
URL_ADMIN_DEFAULT_DEFAULT_CLIENT_SCOPE = URL_ADMIN_DEFAULT_DEFAULT_CLIENT_SCOPES + "/{id}"
URL_ADMIN_DEFAULT_OPTIONAL_CLIENT_SCOPES = URL_ADMIN_REALM + "/default-optional-client-scopes"
URL_ADMIN_DEFAULT_OPTIONAL_CLIENT_SCOPE = URL_ADMIN_DEFAULT_OPTIONAL_CLIENT_SCOPES + "/{id}"
URL_ADMIN_FLOWS = "admin/realms/{realm-name}/authentication/flows"
URL_ADMIN_FLOW = URL_ADMIN_FLOWS + "/{id}"
URL_ADMIN_FLOWS_ALIAS = "admin/realms/{realm-name}/authentication/flows/{flow-id}"
URL_ADMIN_FLOWS_COPY = "admin/realms/{realm-name}/authentication/flows/{flow-alias}/copy"
URL_ADMIN_FLOWS_EXECUTIONS = "admin/realms/{realm-name}/authentication/flows/{flow-alias}/executions"
URL_ADMIN_FLOWS_EXECUTIONS_EXEUCUTION = "admin/realms/{realm-name}/authentication/flows/{flow-alias}/executions/execution"
URL_ADMIN_FLOWS_EXECUTION = "admin/realms/{realm-name}/authentication/executions/{id}"
URL_ADMIN_FLOWS_EXECUTIONS_EXECUTION = "admin/realms/{realm-name}/authentication/flows/{flow-alias}/executions/execution"
URL_ADMIN_FLOWS_EXECUTIONS_FLOW = "admin/realms/{realm-name}/authentication/flows/{flow-alias}/executions/flow"
URL_ADMIN_AUTHENTICATOR_CONFIG = "admin/realms/{realm-name}/authentication/config/{id}"

Loading…
Cancel
Save