Browse Source

Merge branch 'master' into user-id-on-creation

master
Guillaume Troupel 5 years ago
committed by GitHub
parent
commit
aafe2babd5
No known key found for this signature in database GPG Key ID: 4AEE18F83AFDEB23
  1. 38
      Pipfile.lock
  2. 4
      docs/source/conf.py
  3. 16
      docs/source/index.rst
  4. 26
      keycloak/__init__.py
  5. 26
      keycloak/authorization/__init__.py
  6. 12
      keycloak/connection.py
  7. 361
      keycloak/keycloak_admin.py
  8. 9
      keycloak/keycloak_openid.py
  9. 46
      keycloak/tests/test_connection.py
  10. 2
      setup.py

38
Pipfile.lock

@ -1,7 +1,7 @@
{ {
"_meta": { "_meta": {
"hash": { "hash": {
"sha256": "2e38b123d04c65ce270c4f49048a74068545017ba69af6daf4612a5f43f64014"
"sha256": "8c12705e89c665da92fc69ef0d312a9ca313703c839c15d18fcc833dcb87d7f7"
}, },
"pipfile-spec": 6, "pipfile-spec": 6,
"requires": { "requires": {
@ -18,10 +18,10 @@
"default": { "default": {
"certifi": { "certifi": {
"hashes": [ "hashes": [
"sha256:376690d6f16d32f9d1fe8932551d80b23e9d393a8578c5633a2ed39a64861638",
"sha256:456048c7e371c089d0a77a5212fb37a2c2dce1e24146e3b7e0261736aaeaa22a"
"sha256:e4f3620cfea4f83eedc95b24abd9cd56f3c4b146dd0177e83a21b4eb49e21e50",
"sha256:fd7c7c74727ddcf00e9acd26bba8da604ffec95bf1c2144e67aff7a8b50e6cef"
], ],
"version": "==2018.8.24"
"version": "==2019.9.11"
}, },
"chardet": { "chardet": {
"hashes": [ "hashes": [
@ -32,16 +32,17 @@
}, },
"ecdsa": { "ecdsa": {
"hashes": [ "hashes": [
"sha256:40d002cf360d0e035cf2cb985e1308d41aaa087cbfc135b2dc2d844296ea546c",
"sha256:64cf1ee26d1cde3c73c6d7d107f835fed7c6a2904aef9eac223d57ad800c43fa"
"sha256:163c80b064a763ea733870feb96f9dd9b92216cfcacd374837af18e4e8ec3d4d",
"sha256:9814e700890991abeceeb2242586024d4758c8fc18445b194a49bd62d85861db"
], ],
"version": "==0.13"
"index": "pypi",
"version": "==0.13.3"
}, },
"future": { "future": {
"hashes": [ "hashes": [
"sha256:e39ced1ab767b5936646cedba8bcce582398233d6a627067d4c6a454c90cfedb"
"sha256:6142ef79e2416e432931d527452a1cab3aa4a754a0a53d25b2589f79e1106f34"
], ],
"version": "==0.16.0"
"version": "==0.18.0"
}, },
"httmock": { "httmock": {
"hashes": [ "hashes": [
@ -59,10 +60,10 @@
}, },
"pyasn1": { "pyasn1": {
"hashes": [ "hashes": [
"sha256:b9d3abc5031e61927c82d4d96c1cec1e55676c1a991623cfed28faea73cdd7ca",
"sha256:f58f2a3d12fd754aa123e9fa74fb7345333000a035f3921dbdaa08597aa53137"
"sha256:62cdade8b5530f0b185e09855dd422bc05c0bbff6b72ff61381c09dac7befd8c",
"sha256:a9495356ca1d66ed197a0f72b41eb1823cf7ea8b5bd07191673e8147aecf8604"
], ],
"version": "==0.4.4"
"version": "==0.4.7"
}, },
"python-jose": { "python-jose": {
"hashes": [ "hashes": [
@ -82,24 +83,23 @@
}, },
"rsa": { "rsa": {
"hashes": [ "hashes": [
"sha256:25df4e10c263fb88b5ace923dd84bf9aa7f5019687b5e55382ffcdb8bede9db5",
"sha256:43f682fea81c452c98d09fc316aae12de6d30c4b5c84226642cf8f8fd1c93abd"
"sha256:14ba45700ff1ec9eeb206a2ce76b32814958a98e372006c8fb76ba820211be66",
"sha256:1a836406405730121ae9823e19c6e806c62bbad73f890574fff50efa4122c487"
], ],
"version": "==3.4.2"
"version": "==4.0"
}, },
"six": { "six": {
"hashes": [ "hashes": [
"sha256:70e8a77beed4562e7f14fe23a786b54f6296e34344c23bc42f07b15018ff98e9",
"sha256:832dc0e10feb1aa2c68dcc57dbb658f1c7e65b9b61af69048abc87a2db00a0eb"
"sha256:3350809f0555b11f552448330d0b52d5f24c91a322ea4a15ef22629740f3761c",
"sha256:d16a0141ec1a18405cd4ce8b4613101da75da0e9a7aec5bdd4fa804d0e0eba73"
], ],
"version": "==1.11.0"
"version": "==1.12.0"
}, },
"urllib3": { "urllib3": {
"hashes": [ "hashes": [
"sha256:a68ac5e15e76e7e5dd2b8f94007233e01effe3e50e8daddf69acfd81cb686baf", "sha256:a68ac5e15e76e7e5dd2b8f94007233e01effe3e50e8daddf69acfd81cb686baf",
"sha256:b5725a0bd4ba422ab0e66e89e030c806576753ea3ee08554382c14e685d117b5" "sha256:b5725a0bd4ba422ab0e66e89e030c806576753ea3ee08554382c14e685d117b5"
], ],
"markers": "python_version != '3.2.*' and python_version != '3.3.*' and python_version >= '2.6' and python_version < '4' and python_version != '3.1.*' and python_version != '3.0.*'",
"version": "==1.23" "version": "==1.23"
} }
}, },

4
docs/source/conf.py

@ -60,9 +60,9 @@ author = 'Marcos Pereira'
# built documents. # built documents.
# #
# The short X.Y version. # The short X.Y version.
version = '0.17.4'
version = '0.17.6'
# The full version, including alpha/beta/rc tags. # The full version, including alpha/beta/rc tags.
release = '0.17.4'
release = '0.17.6'
# The language for content autogenerated by Sphinx. Refer to documentation # The language for content autogenerated by Sphinx. Refer to documentation
# for a list of supported languages. # for a list of supported languages.

16
docs/source/index.rst

@ -92,6 +92,14 @@ Main methods::
client_secret_key="secret", client_secret_key="secret",
verify=True) verify=True)
# Optionally, you can pass custom headers that will be added to all HTTP calls
# keycloak_openid = KeycloakOpenID(server_url="http://localhost:8080/auth/",
# client_id="example_client",
# realm_name="example_realm",
# client_secret_key="secret",
# verify=True,
# custom_headers={'CustomHeader': 'value'})
# Get WellKnow # Get WellKnow
config_well_know = keycloak_openid.well_know() config_well_know = keycloak_openid.well_know()
@ -143,6 +151,14 @@ Main methods::
realm_name="example_realm", realm_name="example_realm",
verify=True) verify=True)
# Optionally, you can pass custom headers that will be added to all HTTP calls
#keycloak_admin = KeycloakAdmin(server_url="http://localhost:8080/auth/",
# username='example-admin',
# password='secret',
# realm_name="example_realm",
# verify=True,
# custom_headers={'CustomHeader': 'value'})
# Add user # Add user
new_user = keycloak_admin.create_user({"email": "example@example.com", new_user = keycloak_admin.create_user({"email": "example@example.com",
"username": "example@example.com", "username": "example@example.com",

26
keycloak/__init__.py

@ -1,19 +1,25 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# #
# The MIT License (MIT)
#
# Copyright (C) 2017 Marcos Pereira <marcospereira.mpj@gmail.com> # Copyright (C) 2017 Marcos Pereira <marcospereira.mpj@gmail.com>
# #
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# Permission is hereby granted, free of charge, to any person obtaining a copy of
# this software and associated documentation files (the "Software"), to deal in
# the Software without restriction, including without limitation the rights to
# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
# the Software, and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
# #
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
# #
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
# FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
# COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
# IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
from .keycloak_admin import * from .keycloak_admin import *
from .keycloak_openid import * from .keycloak_openid import *

26
keycloak/authorization/__init__.py

@ -1,19 +1,25 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
# #
# The MIT License (MIT)
#
# Copyright (C) 2017 Marcos Pereira <marcospereira.mpj@gmail.com> # Copyright (C) 2017 Marcos Pereira <marcospereira.mpj@gmail.com>
# #
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
# Permission is hereby granted, free of charge, to any person obtaining a copy of
# this software and associated documentation files (the "Software"), to deal in
# the Software without restriction, including without limitation the rights to
# use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of
# the Software, and to permit persons to whom the Software is furnished to do so,
# subject to the following conditions:
# #
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
# #
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS
# FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR
# COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER
# IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN
# CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
import ast import ast
import json import json

12
keycloak/connection.py

@ -27,6 +27,7 @@ except ImportError:
from urlparse import urljoin from urlparse import urljoin
import requests import requests
from requests.adapters import HTTPAdapter
from .exceptions import (KeycloakConnectionError) from .exceptions import (KeycloakConnectionError)
@ -47,6 +48,17 @@ class ConnectionManager(object):
self._verify = verify self._verify = verify
self._s = requests.Session() self._s = requests.Session()
# retry once to reset connection with Keycloak after tomcat's ConnectionTimeout
# see https://github.com/marcospereirampj/python-keycloak/issues/36
for protocol in ('https://', 'http://'):
adapter = HTTPAdapter(max_retries=1)
# adds POST to retry whitelist
method_whitelist = set(adapter.max_retries.method_whitelist)
method_whitelist.add('POST')
adapter.max_retries.method_whitelist = frozenset(method_whitelist)
self._s.mount(protocol, adapter)
@property @property
def base_url(self): def base_url(self):
""" Return base url in use for requests to the server. """ """ Return base url in use for requests to the server. """

361
keycloak/keycloak_admin.py

@ -25,6 +25,8 @@
# internal Keycloak server ID, usually a uuid string # internal Keycloak server ID, usually a uuid string
import json import json
from builtins import isinstance
from typing import List, Iterable
from .connection import ConnectionManager from .connection import ConnectionManager
from .exceptions import raise_error_from_response, KeycloakGetError from .exceptions import raise_error_from_response, KeycloakGetError
@ -45,7 +47,21 @@ class KeycloakAdmin:
PAGE_SIZE = 100 PAGE_SIZE = 100
def __init__(self, server_url, username, password, realm_name='master', client_id='admin-cli', verify=True, client_secret_key=None):
_server_url = None
_username = None
_password = None
_realm_name = None
_client_id = None
_verify = None
_client_secret_key = None
_auto_refresh_token = None
_connection = None
_token = None
_custom_headers = None
_user_realm_name = None
def __init__(self, server_url, username, password, realm_name='master', client_id='admin-cli', verify=True,
client_secret_key=None, custom_headers=None, user_realm_name=None, auto_refresh_token=None):
""" """
:param server_url: Keycloak server url :param server_url: Keycloak server url
@ -55,25 +71,30 @@ class KeycloakAdmin:
:param client_id: client id :param client_id: client id
:param verify: True if want check connection SSL :param verify: True if want check connection SSL
:param client_secret_key: client secret key :param client_secret_key: client secret key
"""
self._username = username
self._password = password
self._client_id = client_id
self._realm_name = realm_name
:param custom_headers: dict of custom header to pass to each HTML request
:param auto_refresh_token: list of methods that allows automatic token refresh. ex: ['get', 'put', 'post', 'delete']
"""
self.server_url = server_url
self.username = username
self.password = password
self.realm_name = realm_name
self.client_id = client_id
self.verify = verify
self.client_secret_key = client_secret_key
self.auto_refresh_token = auto_refresh_token or []
self.user_realm_name = user_realm_name
self.custom_headers = custom_headers
# Get token Admin # Get token Admin
keycloak_openid = KeycloakOpenID(server_url=server_url, client_id=client_id, realm_name=realm_name,
verify=verify, client_secret_key=client_secret_key)
self.get_token()
grant_type = ["password"]
if client_secret_key:
grant_type = ["client_credentials"]
self._token = keycloak_openid.token(username, password, grant_type=grant_type)
self._connection = ConnectionManager(base_url=server_url,
headers={'Authorization': 'Bearer ' + self.token.get('access_token'),
'Content-Type': 'application/json'},
timeout=60,
verify=verify)
@property
def server_url(self):
return self._server_url
@server_url.setter
def server_url(self, value):
self._server_url = value
@property @property
def realm_name(self): def realm_name(self):
@ -99,6 +120,22 @@ class KeycloakAdmin:
def client_id(self, value): def client_id(self, value):
self._client_id = value self._client_id = value
@property
def client_secret_key(self):
return self._client_secret_key
@client_secret_key.setter
def client_secret_key(self, value):
self._client_secret_key = value
@property
def verify(self):
return self._verify
@verify.setter
def verify(self, value):
self._verify = value
@property @property
def username(self): def username(self):
return self._username return self._username
@ -123,6 +160,36 @@ class KeycloakAdmin:
def token(self, value): def token(self, value):
self._token = value self._token = value
@property
def auto_refresh_token(self):
return self._auto_refresh_token
@property
def user_realm_name(self):
return self._user_realm_name
@user_realm_name.setter
def user_realm_name(self, value):
self._user_realm_name = value
@property
def custom_headers(self):
return self._custom_headers
@custom_headers.setter
def custom_headers(self, value):
self._custom_headers = value
@auto_refresh_token.setter
def auto_refresh_token(self, value):
allowed_methods = {'get', 'post', 'put', 'delete'}
if not isinstance(value, Iterable):
raise TypeError('Expected a list of strings among {allowed}'.format(allowed=allowed_methods))
if not all(method in allowed_methods for method in value):
raise TypeError('Unexpected method in auto_refresh_token, accepted methods are {allowed}'.format(allowed=allowed_methods))
self._auto_refresh_token = value
def __fetch_all(self, url, query=None): def __fetch_all(self, url, query=None):
'''Wrapper function to paginate GET requests '''Wrapper function to paginate GET requests
@ -144,7 +211,7 @@ class KeycloakAdmin:
while True: while True:
query['first'] = page*self.PAGE_SIZE query['first'] = page*self.PAGE_SIZE
partial_results = raise_error_from_response( partial_results = raise_error_from_response(
self.connection.raw_get(url, **query),
self.raw_get(url, **query),
KeycloakGetError) KeycloakGetError)
if not partial_results: if not partial_results:
break break
@ -164,8 +231,8 @@ class KeycloakAdmin:
:return: RealmRepresentation :return: RealmRepresentation
""" """
data_raw = self.connection.raw_post(URL_ADMIN_REALMS,
data=json.dumps(payload))
data_raw = self.raw_post(URL_ADMIN_REALMS,
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201)
def get_realms(self): def get_realms(self):
@ -174,22 +241,22 @@ class KeycloakAdmin:
:return: realms list :return: realms list
""" """
data_raw = self.connection.raw_get(URL_ADMIN_REALMS)
data_raw = self.raw_get(URL_ADMIN_REALMS)
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def create_realm(self, payload, skip_exists=False): def create_realm(self, payload, skip_exists=False):
""" """
Create a client
Create a realm
ClientRepresentation: http://www.keycloak.org/docs-api/3.3/rest-api/index.html#_realmrepresentation ClientRepresentation: http://www.keycloak.org/docs-api/3.3/rest-api/index.html#_realmrepresentation
:param skip_exists: Skip if Realm already exist. :param skip_exists: Skip if Realm already exist.
:param payload: RealmRepresentation :param payload: RealmRepresentation
:return: Keycloak server response (UserRepresentation)
:return: Keycloak server response (RealmRepresentation)
""" """
data_raw = self.connection.raw_post(URL_ADMIN_REALMS,
data=json.dumps(payload))
data_raw = self.raw_post(URL_ADMIN_REALMS,
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201, skip_exists=skip_exists) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201, skip_exists=skip_exists)
@ -212,7 +279,7 @@ class KeycloakAdmin:
:return: array IdentityProviderRepresentation :return: array IdentityProviderRepresentation
""" """
params_path = {"realm-name": self.realm_name} params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_ADMIN_IDPS.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_IDPS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def create_user(self, payload): def create_user(self, payload):
@ -233,8 +300,8 @@ class KeycloakAdmin:
if exists is not None: if exists is not None:
return str(exists) return str(exists)
data_raw = self.connection.raw_post(URL_ADMIN_USERS.format(**params_path),
data=json.dumps(payload))
data_raw = self.raw_post(URL_ADMIN_USERS.format(**params_path),
data=json.dumps(payload))
raise_error_from_response(data_raw, KeycloakGetError, expected_code=201) raise_error_from_response(data_raw, KeycloakGetError, expected_code=201)
_last_slash_idx = data_raw.headers['Location'].rindex('/') _last_slash_idx = data_raw.headers['Location'].rindex('/')
return data_raw.headers['Location'][_last_slash_idx + 1:] return data_raw.headers['Location'][_last_slash_idx + 1:]
@ -246,7 +313,7 @@ class KeycloakAdmin:
:return: counter :return: counter
""" """
params_path = {"realm-name": self.realm_name} params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_ADMIN_USERS_COUNT.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_USERS_COUNT.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_user_id(self, username): def get_user_id(self, username):
@ -276,7 +343,7 @@ class KeycloakAdmin:
:return: UserRepresentation :return: UserRepresentation
""" """
params_path = {"realm-name": self.realm_name, "id": user_id} params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.connection.raw_get(URL_ADMIN_USER.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_USER.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_user_groups(self, user_id): def get_user_groups(self, user_id):
@ -288,7 +355,7 @@ class KeycloakAdmin:
:return: user groups list :return: user groups list
""" """
params_path = {"realm-name": self.realm_name, "id": user_id} params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.connection.raw_get(URL_ADMIN_USER_GROUPS.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_USER_GROUPS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def update_user(self, user_id, payload): def update_user(self, user_id, payload):
@ -301,8 +368,8 @@ class KeycloakAdmin:
:return: Http response :return: Http response
""" """
params_path = {"realm-name": self.realm_name, "id": user_id} params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.connection.raw_put(URL_ADMIN_USER.format(**params_path),
data=json.dumps(payload))
data_raw = self.raw_put(URL_ADMIN_USER.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def delete_user(self, user_id): def delete_user(self, user_id):
@ -314,7 +381,7 @@ class KeycloakAdmin:
:return: Http response :return: Http response
""" """
params_path = {"realm-name": self.realm_name, "id": user_id} params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.connection.raw_delete(URL_ADMIN_USER.format(**params_path))
data_raw = self.raw_delete(URL_ADMIN_USER.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def set_user_password(self, user_id, password, temporary=True): def set_user_password(self, user_id, password, temporary=True):
@ -333,8 +400,8 @@ class KeycloakAdmin:
""" """
payload = {"type": "password", "temporary": temporary, "value": password} payload = {"type": "password", "temporary": temporary, "value": password}
params_path = {"realm-name": self.realm_name, "id": user_id} params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.connection.raw_put(URL_ADMIN_RESET_PASSWORD.format(**params_path),
data=json.dumps(payload))
data_raw = self.raw_put(URL_ADMIN_RESET_PASSWORD.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def consents_user(self, user_id): def consents_user(self, user_id):
@ -346,7 +413,7 @@ class KeycloakAdmin:
:return: consents :return: consents
""" """
params_path = {"realm-name": self.realm_name, "id": user_id} params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.connection.raw_get(URL_ADMIN_USER_CONSENTS.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_USER_CONSENTS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def send_update_account(self, user_id, payload, client_id=None, lifespan=None, redirect_uri=None): def send_update_account(self, user_id, payload, client_id=None, lifespan=None, redirect_uri=None):
@ -364,8 +431,8 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": user_id} params_path = {"realm-name": self.realm_name, "id": user_id}
params_query = {"client_id": client_id, "lifespan": lifespan, "redirect_uri": redirect_uri} params_query = {"client_id": client_id, "lifespan": lifespan, "redirect_uri": redirect_uri}
data_raw = self.connection.raw_put(URL_ADMIN_SEND_UPDATE_ACCOUNT.format(**params_path),
data=payload, **params_query)
data_raw = self.raw_put(URL_ADMIN_SEND_UPDATE_ACCOUNT.format(**params_path),
data=payload, **params_query)
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def send_verify_email(self, user_id, client_id=None, redirect_uri=None): def send_verify_email(self, user_id, client_id=None, redirect_uri=None):
@ -381,8 +448,8 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": user_id} params_path = {"realm-name": self.realm_name, "id": user_id}
params_query = {"client_id": client_id, "redirect_uri": redirect_uri} params_query = {"client_id": client_id, "redirect_uri": redirect_uri}
data_raw = self.connection.raw_put(URL_ADMIN_SEND_VERIFY_EMAIL.format(**params_path),
data={}, **params_query)
data_raw = self.raw_put(URL_ADMIN_SEND_VERIFY_EMAIL.format(**params_path),
data={}, **params_query)
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_sessions(self, user_id): def get_sessions(self, user_id):
@ -397,7 +464,7 @@ class KeycloakAdmin:
:return: UserSessionRepresentation :return: UserSessionRepresentation
""" """
params_path = {"realm-name": self.realm_name, "id": user_id} params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.connection.raw_get(URL_ADMIN_GET_SESSIONS.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_GET_SESSIONS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_server_info(self): def get_server_info(self):
@ -409,7 +476,7 @@ class KeycloakAdmin:
:return: ServerInfoRepresentation :return: ServerInfoRepresentation
""" """
data_raw = self.connection.raw_get(URL_ADMIN_SERVER_INFO)
data_raw = self.raw_get(URL_ADMIN_SERVER_INFO)
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_groups(self): def get_groups(self):
@ -434,7 +501,7 @@ class KeycloakAdmin:
:return: Keycloak server response (GroupRepresentation) :return: Keycloak server response (GroupRepresentation)
""" """
params_path = {"realm-name": self.realm_name, "id": group_id} params_path = {"realm-name": self.realm_name, "id": group_id}
data_raw = self.connection.raw_get(URL_ADMIN_GROUP.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_GROUP.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_subgroups(self, group, path): def get_subgroups(self, group, path):
@ -517,12 +584,12 @@ class KeycloakAdmin:
if parent is None: if parent is None:
params_path = {"realm-name": self.realm_name} params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_post(URL_ADMIN_GROUPS.format(**params_path),
data=json.dumps(payload))
data_raw = self.raw_post(URL_ADMIN_GROUPS.format(**params_path),
data=json.dumps(payload))
else: else:
params_path = {"realm-name": self.realm_name, "id": parent, } params_path = {"realm-name": self.realm_name, "id": parent, }
data_raw = self.connection.raw_post(URL_ADMIN_GROUP_CHILD.format(**params_path),
data=json.dumps(payload))
data_raw = self.raw_post(URL_ADMIN_GROUP_CHILD.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201, skip_exists=skip_exists) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201, skip_exists=skip_exists)
@ -540,8 +607,8 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": group_id} params_path = {"realm-name": self.realm_name, "id": group_id}
data_raw = self.connection.raw_put(URL_ADMIN_GROUP.format(**params_path),
data=json.dumps(payload))
data_raw = self.raw_put(URL_ADMIN_GROUP.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def group_set_permissions(self, group_id, enabled=True): def group_set_permissions(self, group_id, enabled=True):
@ -554,8 +621,8 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": group_id} params_path = {"realm-name": self.realm_name, "id": group_id}
data_raw = self.connection.raw_put(URL_ADMIN_GROUP_PERMISSIONS.format(**params_path),
data=json.dumps({"enabled": enabled}))
data_raw = self.raw_put(URL_ADMIN_GROUP_PERMISSIONS.format(**params_path),
data=json.dumps({"enabled": enabled}))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def group_user_add(self, user_id, group_id): def group_user_add(self, user_id, group_id):
@ -569,7 +636,7 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": user_id, "group-id": group_id} params_path = {"realm-name": self.realm_name, "id": user_id, "group-id": group_id}
data_raw = self.connection.raw_put(URL_ADMIN_USER_GROUP.format(**params_path), data=None)
data_raw = self.raw_put(URL_ADMIN_USER_GROUP.format(**params_path), data=None)
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def group_user_remove(self, user_id, group_id): def group_user_remove(self, user_id, group_id):
@ -583,7 +650,7 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": user_id, "group-id": group_id} params_path = {"realm-name": self.realm_name, "id": user_id, "group-id": group_id}
data_raw = self.connection.raw_delete(URL_ADMIN_USER_GROUP.format(**params_path))
data_raw = self.raw_delete(URL_ADMIN_USER_GROUP.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def delete_group(self, group_id): def delete_group(self, group_id):
@ -595,7 +662,7 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": group_id} params_path = {"realm-name": self.realm_name, "id": group_id}
data_raw = self.connection.raw_delete(URL_ADMIN_GROUP.format(**params_path))
data_raw = self.raw_delete(URL_ADMIN_GROUP.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def get_clients(self): def get_clients(self):
@ -609,7 +676,7 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name} params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_ADMIN_CLIENTS.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_CLIENTS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_client(self, client_id): def get_client(self, client_id):
@ -624,7 +691,7 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": client_id} params_path = {"realm-name": self.realm_name, "id": client_id}
data_raw = self.connection.raw_get(URL_ADMIN_CLIENT.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_CLIENT.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_client_id(self, client_name): def get_client_id(self, client_name):
@ -655,7 +722,7 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": client_id} params_path = {"realm-name": self.realm_name, "id": client_id}
data_raw = self.connection.raw_get(URL_ADMIN_CLIENT_AUTHZ_SETTINGS.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_CLIENT_AUTHZ_SETTINGS.format(**params_path))
return data_raw return data_raw
def get_client_authz_resources(self, client_id): def get_client_authz_resources(self, client_id):
@ -668,7 +735,7 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": client_id} params_path = {"realm-name": self.realm_name, "id": client_id}
data_raw = self.connection.raw_get(URL_ADMIN_CLIENT_AUTHZ_RESOURCES.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_CLIENT_AUTHZ_RESOURCES.format(**params_path))
return data_raw return data_raw
def create_client(self, payload, skip_exists=False): def create_client(self, payload, skip_exists=False):
@ -683,10 +750,24 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name} params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_post(URL_ADMIN_CLIENTS.format(**params_path),
data=json.dumps(payload))
data_raw = self.raw_post(URL_ADMIN_CLIENTS.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201, skip_exists=skip_exists) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201, skip_exists=skip_exists)
def update_client(self, client_id, payload):
"""
Update a client
:param client_id: Client id
:param payload: ClientRepresentation
:return: Http response
"""
params_path = {"realm-name": self.realm_name, "id": client_id}
data_raw = self.connection.raw_put(URL_ADMIN_CLIENT.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def delete_client(self, client_id): def delete_client(self, client_id):
""" """
Get representation of the client Get representation of the client
@ -699,7 +780,7 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": client_id} params_path = {"realm-name": self.realm_name, "id": client_id}
data_raw = self.connection.raw_delete(URL_ADMIN_CLIENT.format(**params_path))
data_raw = self.raw_delete(URL_ADMIN_CLIENT.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def get_realm_roles(self): def get_realm_roles(self):
@ -713,7 +794,7 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name} params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_ADMIN_REALM_ROLES.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_REALM_ROLES.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_client_roles(self, client_id): def get_client_roles(self, client_id):
@ -729,7 +810,7 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": client_id} params_path = {"realm-name": self.realm_name, "id": client_id}
data_raw = self.connection.raw_get(URL_ADMIN_CLIENT_ROLES.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_CLIENT_ROLES.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_client_role(self, client_id, role_name): def get_client_role(self, client_id, role_name):
@ -746,7 +827,7 @@ class KeycloakAdmin:
:return: role_id :return: role_id
""" """
params_path = {"realm-name": self.realm_name, "id": client_id, "role-name": role_name} params_path = {"realm-name": self.realm_name, "id": client_id, "role-name": role_name}
data_raw = self.connection.raw_get(URL_ADMIN_CLIENT_ROLE.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_CLIENT_ROLE.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_client_role_id(self, client_id, role_name): def get_client_role_id(self, client_id, role_name):
@ -780,8 +861,8 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": client_role_id} params_path = {"realm-name": self.realm_name, "id": client_role_id}
data_raw = self.connection.raw_post(URL_ADMIN_CLIENT_ROLES.format(**params_path),
data=json.dumps(payload))
data_raw = self.raw_post(URL_ADMIN_CLIENT_ROLES.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201, skip_exists=skip_exists) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201, skip_exists=skip_exists)
def delete_client_role(self, client_role_id, role_name): def delete_client_role(self, client_role_id, role_name):
@ -795,7 +876,7 @@ class KeycloakAdmin:
:param role_name: roles name (not id!) :param role_name: roles name (not id!)
""" """
params_path = {"realm-name": self.realm_name, "id": client_role_id, "role-name": role_name} params_path = {"realm-name": self.realm_name, "id": client_role_id, "role-name": role_name}
data_raw = self.connection.raw_delete(URL_ADMIN_CLIENT_ROLE.format(**params_path))
data_raw = self.raw_delete(URL_ADMIN_CLIENT_ROLE.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def assign_client_role(self, user_id, client_id, roles): def assign_client_role(self, user_id, client_id, roles):
@ -811,10 +892,25 @@ class KeycloakAdmin:
payload = roles if isinstance(roles, list) else [roles] payload = roles if isinstance(roles, list) else [roles]
params_path = {"realm-name": self.realm_name, "id": user_id, "client-id": client_id} params_path = {"realm-name": self.realm_name, "id": user_id, "client-id": client_id}
data_raw = self.connection.raw_post(URL_ADMIN_USER_CLIENT_ROLES.format(**params_path),
data=json.dumps(payload))
data_raw = self.raw_post(URL_ADMIN_USER_CLIENT_ROLES.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def create_realm_role(self, payload, skip_exists=False):
"""
Create a new role for the realm or client
:param realm: realm name (not id)
:param rep: RoleRepresentation https://www.keycloak.org/docs-api/5.0/rest-api/index.html#_rolerepresentation
:return Keycloak server response
"""
params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_post(URL_ADMIN_REALM_ROLES.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201, skip_exists=skip_exists)
def assign_realm_roles(self, user_id, client_id, roles): def assign_realm_roles(self, user_id, client_id, roles):
""" """
Assign realm roles to a user Assign realm roles to a user
@ -828,8 +924,8 @@ class KeycloakAdmin:
payload = roles if isinstance(roles, list) else [roles] payload = roles if isinstance(roles, list) else [roles]
params_path = {"realm-name": self.realm_name, "id": user_id} params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.connection.raw_post(URL_ADMIN_USER_REALM_ROLES.format(**params_path),
data=json.dumps(payload))
data_raw = self.raw_post(URL_ADMIN_USER_REALM_ROLES.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def get_client_roles_of_user(self, user_id, client_id): def get_client_roles_of_user(self, user_id, client_id):
@ -864,7 +960,7 @@ class KeycloakAdmin:
def _get_client_roles_of_user(self, client_level_role_mapping_url, user_id, client_id): def _get_client_roles_of_user(self, client_level_role_mapping_url, user_id, client_id):
params_path = {"realm-name": self.realm_name, "id": user_id, "client-id": client_id} params_path = {"realm-name": self.realm_name, "id": user_id, "client-id": client_id}
data_raw = self.connection.raw_get(client_level_role_mapping_url.format(**params_path))
data_raw = self.raw_get(client_level_role_mapping_url.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def delete_client_roles_of_user(self, user_id, client_id, roles): def delete_client_roles_of_user(self, user_id, client_id, roles):
@ -879,8 +975,8 @@ class KeycloakAdmin:
""" """
payload = roles if isinstance(roles, list) else [roles] payload = roles if isinstance(roles, list) else [roles]
params_path = {"realm-name": self.realm_name, "id": user_id, "client-id": client_id} params_path = {"realm-name": self.realm_name, "id": user_id, "client-id": client_id}
data_raw = self.connection.raw_delete(URL_ADMIN_USER_CLIENT_ROLES.format(**params_path),
data=json.dumps(payload))
data_raw = self.raw_delete(URL_ADMIN_USER_CLIENT_ROLES.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def get_authentication_flows(self): def get_authentication_flows(self):
@ -893,7 +989,7 @@ class KeycloakAdmin:
:return: Keycloak server response (AuthenticationFlowRepresentation) :return: Keycloak server response (AuthenticationFlowRepresentation)
""" """
params_path = {"realm-name": self.realm_name} params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_ADMIN_FLOWS.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_FLOWS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def create_authentication_flow(self, payload, skip_exists=False): def create_authentication_flow(self, payload, skip_exists=False):
@ -908,8 +1004,8 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name} params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_post(URL_ADMIN_FLOWS.format(**params_path),
data=payload)
data_raw = self.raw_post(URL_ADMIN_FLOWS.format(**params_path),
data=payload)
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201, skip_exists=skip_exists) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201, skip_exists=skip_exists)
def get_authentication_flow_executions(self, flow_alias): def get_authentication_flow_executions(self, flow_alias):
@ -919,7 +1015,7 @@ class KeycloakAdmin:
:return: Response(json) :return: Response(json)
""" """
params_path = {"realm-name": self.realm_name, "flow-alias": flow_alias} params_path = {"realm-name": self.realm_name, "flow-alias": flow_alias}
data_raw = self.connection.raw_get(URL_ADMIN_FLOWS_EXECUTIONS.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_FLOWS_EXECUTIONS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def update_authentication_flow_executions(self, payload, flow_alias): def update_authentication_flow_executions(self, payload, flow_alias):
@ -934,8 +1030,8 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "flow-alias": flow_alias} params_path = {"realm-name": self.realm_name, "flow-alias": flow_alias}
data_raw = self.connection.raw_put(URL_ADMIN_FLOWS_EXECUTIONS.format(**params_path),
data=payload)
data_raw = self.raw_put(URL_ADMIN_FLOWS_EXECUTIONS.format(**params_path),
data=payload)
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def sync_users(self, storage_id, action): def sync_users(self, storage_id, action):
@ -950,8 +1046,8 @@ class KeycloakAdmin:
params_query = {"action": action} params_query = {"action": action}
params_path = {"realm-name": self.realm_name, "id": storage_id} params_path = {"realm-name": self.realm_name, "id": storage_id}
data_raw = self.connection.raw_post(URL_ADMIN_USER_STORAGE.format(**params_path),
data=json.dumps(data), **params_query)
data_raw = self.raw_post(URL_ADMIN_USER_STORAGE.format(**params_path),
data=json.dumps(data), **params_query)
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_client_scopes(self): def get_client_scopes(self):
@ -963,7 +1059,7 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name} params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_ADMIN_CLIENT_SCOPES.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_CLIENT_SCOPES.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def get_client_scope(self, client_scope_id): def get_client_scope(self, client_scope_id):
@ -975,7 +1071,7 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "scope-id": client_scope_id} params_path = {"realm-name": self.realm_name, "scope-id": client_scope_id}
data_raw = self.connection.raw_get(URL_ADMIN_CLIENT_SCOPE.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_CLIENT_SCOPE.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
@ -990,7 +1086,8 @@ class KeycloakAdmin:
params_path = {"realm-name": self.realm_name, "scope-id": client_scope_id} params_path = {"realm-name": self.realm_name, "scope-id": client_scope_id}
data_raw = self.connection.raw_post(URL_ADMIN_CLIENT_SCOPES_ADD_MAPPER.format(**params_path), data=json.dumps(payload))
data_raw = self.raw_post(
URL_ADMIN_CLIENT_SCOPES_ADD_MAPPER.format(**params_path), data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201) return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201)
@ -1005,5 +1102,95 @@ class KeycloakAdmin:
""" """
params_path = {"realm-name": self.realm_name, "id": client_id} params_path = {"realm-name": self.realm_name, "id": client_id}
data_raw = self.connection.raw_get(URL_ADMIN_CLIENT_SECRETS.format(**params_path))
data_raw = self.raw_get(URL_ADMIN_CLIENT_SECRETS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError) return raise_error_from_response(data_raw, KeycloakGetError)
def raw_get(self, *args, **kwargs):
"""
Calls 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.
"""
r = self.connection.raw_get(*args, **kwargs)
if 'get' in self.auto_refresh_token and r.status_code == 401:
self.refresh_token()
return self.connection.raw_get(*args, **kwargs)
return r
def raw_post(self, *args, **kwargs):
"""
Calls 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.
"""
r = self.connection.raw_post(*args, **kwargs)
if 'post' in self.auto_refresh_token and r.status_code == 401:
self.refresh_token()
return self.connection.raw_post(*args, **kwargs)
return r
def raw_put(self, *args, **kwargs):
"""
Calls 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.
"""
r = self.connection.raw_put(*args, **kwargs)
if 'put' in self.auto_refresh_token and r.status_code == 401:
self.refresh_token()
return self.connection.raw_put(*args, **kwargs)
return r
def raw_delete(self, *args, **kwargs):
"""
Calls 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.
"""
r = self.connection.raw_delete(*args, **kwargs)
if 'delete' in self.auto_refresh_token and r.status_code == 401:
self.refresh_token()
return self.connection.raw_delete(*args, **kwargs)
return r
def get_token(self):
self.keycloak_openid = KeycloakOpenID(server_url=self.server_url, client_id=self.client_id,
realm_name=self.user_realm_name or self.realm_name, verify=self.verify,
client_secret_key=self.client_secret_key,
custom_headers=self.custom_headers)
grant_type = ["password"]
if self.client_secret_key:
grant_type = ["client_credentials"]
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'
}
if self.custom_headers is not None:
# merge custom headers to main headers
headers.update(self.custom_headers)
self._connection = ConnectionManager(base_url=self.server_url,
headers=headers,
timeout=60,
verify=self.verify)
def refresh_token(self):
refresh_token = self.token.get('refresh_token')
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:
self.get_token()
else:
raise
self.connection.add_param_headers('Authorization', 'Bearer ' + self.token.get('access_token'))

9
keycloak/keycloak_openid.py

@ -43,7 +43,7 @@ from .urls_patterns import (
class KeycloakOpenID: class KeycloakOpenID:
def __init__(self, server_url, realm_name, client_id, client_secret_key=None, verify=True):
def __init__(self, server_url, realm_name, client_id, client_secret_key=None, verify=True, custom_headers=None):
""" """
:param server_url: Keycloak server url :param server_url: Keycloak server url
@ -51,12 +51,17 @@ class KeycloakOpenID:
:param realm_name: realm name :param realm_name: realm name
:param client_secret_key: client secret key :param client_secret_key: client secret key
:param verify: True if want check connection SSL :param verify: True if want check connection SSL
:param custom_headers: dict of custom header to pass to each HTML request
""" """
self._client_id = client_id self._client_id = client_id
self._client_secret_key = client_secret_key self._client_secret_key = client_secret_key
self._realm_name = realm_name self._realm_name = realm_name
headers = dict()
if custom_headers is not None:
# merge custom headers to main headers
headers.update(custom_headers)
self._connection = ConnectionManager(base_url=server_url, self._connection = ConnectionManager(base_url=server_url,
headers={},
headers=headers,
timeout=60, timeout=60,
verify=verify) verify=verify)

46
keycloak/tests/test_connection.py

@ -14,9 +14,11 @@
# #
# You should have received a copy of the GNU Lesser General Public License # You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>. # along with this program. If not, see <http://www.gnu.org/licenses/>.
from unittest import mock
from httmock import urlmatch, response, HTTMock, all_requests from httmock import urlmatch, response, HTTMock, all_requests
from keycloak import KeycloakAdmin, KeycloakOpenID
from ..connection import ConnectionManager from ..connection import ConnectionManager
try: try:
@ -141,3 +143,47 @@ class TestConnection(unittest.TestCase):
self._conn.add_param_headers("test", "value") self._conn.add_param_headers("test", "value")
self.assertEqual(self._conn.headers, self.assertEqual(self._conn.headers,
{"test": "value"}) {"test": "value"})
def test_KeycloakAdmin_custom_header(self):
class FakeToken:
@staticmethod
def get(string_val):
return "faketoken"
fake_token = FakeToken()
with mock.patch.object(KeycloakOpenID, "__init__", return_value=None) as mock_keycloak_open_id:
with mock.patch("keycloak.keycloak_openid.KeycloakOpenID.token", return_value=fake_token):
with mock.patch("keycloak.connection.ConnectionManager.__init__", return_value=None) as mock_connection_manager:
server_url = "https://localhost/auth/"
username = "admin"
password = "secret"
realm_name = "master"
headers = {
'Custom': 'test-custom-header'
}
KeycloakAdmin(server_url=server_url,
username=username,
password=password,
realm_name=realm_name,
verify=False,
custom_headers=headers)
mock_keycloak_open_id.assert_called_with(server_url=server_url,
realm_name=realm_name,
client_id='admin-cli',
client_secret_key=None,
verify=False,
custom_headers=headers)
expected_header = {'Authorization': 'Bearer faketoken',
'Content-Type': 'application/json',
'Custom': 'test-custom-header'
}
mock_connection_manager.assert_called_with(base_url=server_url,
headers=expected_header,
timeout=60,
verify=False)

2
setup.py

@ -7,7 +7,7 @@ with open("README.md", "r") as fh:
setup( setup(
name='python-keycloak', name='python-keycloak',
version='0.17.5',
version='0.17.6',
url='https://github.com/marcospereirampj/python-keycloak', url='https://github.com/marcospereirampj/python-keycloak',
license='The MIT License', license='The MIT License',
author='Marcos Pereira', author='Marcos Pereira',

Loading…
Cancel
Save