Browse Source

Merge branch 'feature/admin-cli' into develop

pull/12/head
Marcos Pereira 7 years ago
parent
commit
300bf61f92
  1. 1
      .gitignore
  2. 95
      README.md
  3. 4
      docs/source/conf.py
  4. 96
      docs/source/index.rst
  5. 350
      keycloak/__init__.py
  6. 19
      keycloak/connection.py
  7. 304
      keycloak/keycloak_admin.py
  8. 365
      keycloak/keycloak_openid.py
  9. 22
      keycloak/urls_patterns.py
  10. 2
      setup.py

1
.gitignore

@ -102,4 +102,5 @@ ENV/
.idea/
main.py
main2.py
s3air-authz-config.json

95
README.md

@ -44,49 +44,112 @@ The documentation for python-keycloak is available on [readthedocs](http://pytho
## Usage
```python
from keycloak import Keycloak
from keycloak import KeycloakOpenID
# Configure client
keycloak = Keycloak(server_url="http://localhost:8080/auth/",
keycloak_openid = KeycloakOpenID(server_url="http://localhost:8080/auth/",
client_id="example_client",
realm_name="example_realm",
client_secret_key="secret")
# Get WellKnow
config_well_know = keycloak.well_know()
config_well_know = keycloak_openid.well_know()
# Get Token
token = keycloak.token("user", "password")
token = keycloak_openid.token("user", "password")
# Get Userinfo
userinfo = keycloak.userinfo(token['access_token'])
userinfo = keycloak_openid.userinfo(token['access_token'])
# Logout
keycloak.logout(token['refresh_token'])
keycloak_openid.logout(token['refresh_token'])
# Get Certs
certs = keycloak.certs()
certs = keycloak_openid.certs()
# Get RPT (Entitlement)
token = keycloak.token("user", "password")
rpt = keycloak.entitlement(token['access_token'], "resource_id")
token = keycloak_openid.token("user", "password")
rpt = keycloak_openid.entitlement(token['access_token'], "resource_id")
# Instropect RPT
token_rpt_info = keycloak.instropect(keycloak.instropect(token['access_token'], rpt=rpt['rpt'],
token_rpt_info = keycloak_openid.instropect(keycloak_openid.instropect(token['access_token'], rpt=rpt['rpt'],
token_type_hint="requesting_party_token"))
# Introspect Token
token_info = keycloak.introspect(token['access_token']))
token_info = keycloak_openid.introspect(token['access_token']))
# Decode Token
KEYCLOAK_PUBLIC_KEY = "secret"
options = {"verify_signature": True, "verify_aud": True, "exp": True}
token_info = keycloak.decode_token(token['access_token'], key=KEYCLOAK_PUBLIC_KEY, options=options)
token_info = keycloak_openid.decode_token(token['access_token'], key=KEYCLOAK_PUBLIC_KEY, options=options)
# Get permissions by token
token = keycloak.token("user", "password")
keycloak.load_authorization_config("example-authz-config.json")
policies = keycloak.get_policies(token['access_token'], method_token_info='decode', key=KEYCLOAK_PUBLIC_KEY)
permissions = keycloak.get_permissions(token['access_token'], method_token_info='introspect')
token = keycloak_openid.token("user", "password")
keycloak_openid.load_authorization_config("example-authz-config.json")
policies = keycloak_openid.get_policies(token['access_token'], method_token_info='decode', key=KEYCLOAK_PUBLIC_KEY)
permissions = keycloak_openid.get_permissions(token['access_token'], method_token_info='introspect')
# KEYCLOAK ADMIN
from keycloak import KeycloakAdmin
keycloak_admin = KeycloakAdmin(server_url="http://localhost:8080/auth/",
username='example-admin',
password='secret',
realm_name="example_realm")
# Add user
new_user = keycloak_admin.create_user({"email": "example@example.com",
"username": "example@example.com",
"enabled": True,
"firstName": "Example",
"lastName": "Example",
"realmRoles": ["user_default", ],
"attributes": {"example": "1,2,3,3,"}})
# User counter
count_users = keycloak_admin.users_count()
# Get users Returns a list of users, filtered according to query parameters
users = keycloak_admin.get_users({})
# Get User
user = keycloak_admin.get_user("user-id-keycloak")
# Update User
response = keycloak_admin.update_user(user_id="user-id-keycloak",
payload={'firstName': 'Example Update'})
# Delete User
response = keycloak_admin.delete_user(user_id="user-id-keycloak")
# Get consents granted by the user
consents = keycloak_admin.consents_user(user_id="user-id-keycloak")
# Send User Action
response = keycloak_admin.send_update_account(user_id="user-id-keycloak",
payload=json.dumps(['UPDATE_PASSWORD']))
# Send Verify Email
response = keycloak_admin.send_verify_email(user_id="user-id-keycloak")
# Get sessions associated with the user
sessions = keycloak_admin.get_sessions(user_id="user-id-keycloak")
# Get themes, social providers, auth providers, and event listeners available on this server
server_info = keycloak_admin.get_server_info()
# Get clients belonging to the realm Returns a list of clients belonging to the realm
clients = keycloak_admin.get_clients()
# Get representation of the client - id of client (not client-id)
client = keycloak_admin.get_client(client_id='id-client')
# Get all roles for the client
client_roles = keycloak_admin.get_client_role(client_id='id-client')
# Get all roles for the realm or client
realm_roles = keycloak_admin.get_roles()
```

4
docs/source/conf.py

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

96
docs/source/index.rst

@ -71,48 +71,112 @@ Usage
Main methods::
from keycloak import Keycloak
from keycloak import KeycloakOpenID
# Configure client
keycloak = Keycloak(server_url="http://localhost:8080/auth/",
keycloak_openid = KeycloakOpenID(server_url="http://localhost:8080/auth/",
client_id="example_client",
realm_name="example_realm",
client_secret_key="secret")
# Get WellKnow
config_well_know = keycloak.well_know()
config_well_know = keycloak_openid.well_know()
# Get Token
token = keycloak.token("user", "password")
token = keycloak_openid.token("user", "password")
# Get Userinfo
userinfo = keycloak.userinfo(token['access_token'])
userinfo = keycloak_openid.userinfo(token['access_token'])
# Logout
keycloak.logout(token['refresh_token'])
keycloak_openid.logout(token['refresh_token'])
# Get Certs
certs = keycloak.certs()
certs = keycloak_openid.certs()
# Get RPT (Entitlement)
token = keycloak.token("user", "password")
rpt = keycloak.entitlement(token['access_token'], "resource_id")
token = keycloak_openid.token("user", "password")
rpt = keycloak_openid.entitlement(token['access_token'], "resource_id")
# Instropect RPT
token_rpt_info = keycloak.instropect(keycloak.instropect(token['access_token'], rpt=rpt['rpt'],
token_rpt_info = keycloak_openid.instropect(keycloak_openid.instropect(token['access_token'], rpt=rpt['rpt'],
token_type_hint="requesting_party_token"))
# Introspect Token
token_info = keycloak.introspect(token['access_token']))
token_info = keycloak_openid.introspect(token['access_token']))
# Decode Token
KEYCLOAK_PUBLIC_KEY = "secret"
options = {"verify_signature": True, "verify_aud": True, "exp": True}
token_info = keycloak.decode_token(token['access_token'], key=KEYCLOAK_PUBLIC_KEY, options=options)
token_info = keycloak_openid.decode_token(token['access_token'], key=KEYCLOAK_PUBLIC_KEY, options=options)
# Get permissions by token
token = keycloak.token("user", "password")
keycloak.load_authorization_config("example-authz-config.json")
policies = keycloak.get_policies(token['access_token'], method_token_info='decode', key=KEYCLOAK_PUBLIC_KEY)
permissions = keycloak.get_permissions(token['access_token'], method_token_info='introspect')
token = keycloak_openid.token("user", "password")
keycloak_openid.load_authorization_config("example-authz-config.json")
policies = keycloak_openid.get_policies(token['access_token'], method_token_info='decode', key=KEYCLOAK_PUBLIC_KEY)
permissions = keycloak_openid.get_permissions(token['access_token'], method_token_info='introspect')
# KEYCLOAK ADMIN
from keycloak import KeycloakAdmin
keycloak_admin = KeycloakAdmin(server_url="http://localhost:8080/auth/",
username='example-admin',
password='secret',
realm_name="example_realm")
# Add user
new_user = keycloak_admin.create_user({"email": "example@example.com",
"username": "example@example.com",
"enabled": True,
"firstName": "Example",
"lastName": "Example",
"realmRoles": ["user_default", ],
"attributes": {"example": "1,2,3,3,"}})
# User counter
count_users = keycloak_admin.users_count()
# Get users Returns a list of users, filtered according to query parameters
users = keycloak_admin.get_users({})
# Get User
user = keycloak_admin.get_user("user-id-keycloak")
# Update User
response = keycloak_admin.update_user(user_id="user-id-keycloak",
payload={'firstName': 'Example Update'})
# Delete User
response = keycloak_admin.delete_user(user_id="user-id-keycloak")
# Get consents granted by the user
consents = keycloak_admin.consents_user(user_id="user-id-keycloak")
# Send User Action
response = keycloak_admin.send_update_account(user_id="user-id-keycloak",
payload=json.dumps(['UPDATE_PASSWORD']))
# Send Verify Email
response = keycloak_admin.send_verify_email(user_id="user-id-keycloak")
# Get sessions associated with the user
sessions = keycloak_admin.get_sessions(user_id="user-id-keycloak")
# Get themes, social providers, auth providers, and event listeners available on this server
server_info = keycloak_admin.get_server_info()
# Get clients belonging to the realm Returns a list of clients belonging to the realm
clients = keycloak_admin.get_clients()
# Get representation of the client - id of client (not client-id)
client = keycloak_admin.get_client(client_id='id-client')
# Get all roles for the client
client_roles = keycloak_admin.get_client_role(client_id='id-client')
# Get all roles for the realm or client
realm_roles = keycloak_admin.get_roles()

350
keycloak/__init__.py

@ -14,352 +14,6 @@
#
# 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/>.
from .authorization import Authorization
from .exceptions import raise_error_from_response, KeycloakGetError, KeycloakSecretNotFound, \
KeycloakRPTNotFound, KeycloakAuthorizationConfigError, KeycloakInvalidTokenError
from .urls_patterns import (
URL_AUTH,
URL_TOKEN,
URL_USERINFO,
URL_WELL_KNOWN,
URL_LOGOUT,
URL_CERTS,
URL_ENTITLEMENT,
URL_INTROSPECT
)
from .connection import ConnectionManager
from jose import jwt
import json
class Keycloak:
def __init__(self, server_url, client_id, realm_name, client_secret_key=None):
self._client_id = client_id
self._client_secret_key = client_secret_key
self._realm_name = realm_name
self._connection = ConnectionManager(base_url=server_url,
headers={},
timeout=60)
self._authorization = Authorization()
@property
def client_id(self):
return self._client_id
@client_id.setter
def client_id(self, 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 realm_name(self):
return self._realm_name
@realm_name.setter
def realm_name(self, value):
self._realm_name = value
@property
def connection(self):
return self._connection
@connection.setter
def connection(self, value):
self._connection = value
@property
def authorization(self):
return self._authorization
@authorization.setter
def authorization(self, value):
self._authorization = value
def _add_secret_key(self, payload):
"""
Add secret key if exist.
:param payload:
:return:
"""
if self.client_secret_key:
payload.update({"client_secret": self.client_secret_key})
return payload
def _build_name_role(self, role):
"""
:param role:
:return:
"""
return self.client_id + "/" + role
def _token_info(self, token, method_token_info, **kwargs):
"""
:param token:
:param method_token_info:
:param kwargs:
:return:
"""
if method_token_info == 'introspect':
token_info = self.introspect(token)
else:
token_info = self.decode_token(token, **kwargs)
return token_info
def well_know(self):
""" The most important endpoint to understand is the well-known configuration
endpoint. It lists endpoints and other configuration options relevant to
the OpenID Connect implementation in Keycloak.
:return It lists endpoints and other configuration options relevant.
"""
params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_WELL_KNOWN.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def auth_url(self, redirect_uri):
"""
http://openid.net/specs/openid-connect-core-1_0.html#AuthorizationEndpoint
:return:
"""
return NotImplemented
def token(self, username, password, grant_type=["password"]):
"""
The token endpoint is used to obtain tokens. Tokens can either be obtained by
exchanging an authorization code or by supplying credentials directly depending on
what flow is used. The token endpoint is also used to obtain new access tokens
when they expire.
http://openid.net/specs/openid-connect-core-1_0.html#TokenEndpoint
:param username:
:param password:
:param grant_type:
:return:
"""
params_path = {"realm-name": self.realm_name}
payload = {"username": username, "password": password,
"client_id": self.client_id, "grant_type": grant_type}
payload = self._add_secret_key(payload)
data_raw = self.connection.raw_post(URL_TOKEN.format(**params_path),
data=payload)
return raise_error_from_response(data_raw, KeycloakGetError)
def userinfo(self, token):
"""
The userinfo endpoint returns standard claims about the authenticated user,
and is protected by a bearer token.
http://openid.net/specs/openid-connect-core-1_0.html#UserInfo
:param token:
:return:
"""
self.connection.add_param_headers("Authorization", "Bearer " + token)
params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_USERINFO.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def logout(self, refresh_token):
"""
The logout endpoint logs out the authenticated user.
:param refresh_token:
:return:
"""
params_path = {"realm-name": self.realm_name}
payload = {"client_id": self.client_id, "refresh_token": refresh_token}
payload = self._add_secret_key(payload)
data_raw = self.connection.raw_post(URL_LOGOUT.format(**params_path),
data=payload)
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def certs(self):
"""
The certificate endpoint returns the public keys enabled by the realm, encoded as a
JSON Web Key (JWK). Depending on the realm settings there can be one or more keys enabled
for verifying tokens.
https://tools.ietf.org/html/rfc7517
:return:
"""
params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_CERTS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def entitlement(self, token, resource_server_id):
"""
Client applications can use a specific endpoint to obtain a special security token
called a requesting party token (RPT). This token consists of all the entitlements
(or permissions) for a user as a result of the evaluation of the permissions and authorization
policies associated with the resources being requested. With an RPT, client applications can
gain access to protected resources at the resource server.
:return:
"""
self.connection.add_param_headers("Authorization", "Bearer " + token)
params_path = {"realm-name": self.realm_name, "resource-server-id": resource_server_id}
data_raw = self.connection.raw_get(URL_ENTITLEMENT.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def introspect(self, token, rpt=None, token_type_hint=None):
"""
The introspection endpoint is used to retrieve the active state of a token. It is can only be
invoked by confidential clients.
https://tools.ietf.org/html/rfc7662
:param token:
:param rpt:
:param token_type_hint:
:return:
"""
params_path = {"realm-name": self.realm_name}
payload = {"client_id": self.client_id, "token": token}
if token_type_hint == 'requesting_party_token':
if rpt:
payload.update({"token": rpt, "token_type_hint": token_type_hint})
self.connection.add_param_headers("Authorization", "Bearer " + token)
else:
raise KeycloakRPTNotFound("Can't found RPT.")
payload = self._add_secret_key(payload)
data_raw = self.connection.raw_post(URL_INTROSPECT.format(**params_path),
data=payload)
return raise_error_from_response(data_raw, KeycloakGetError)
def decode_token(self, token, key, algorithms=['RS256'], **kwargs):
"""
A JSON Web Key (JWK) is a JavaScript Object Notation (JSON) data
structure that represents a cryptographic key. This specification
also defines a JWK Set JSON data structure that represents a set of
JWKs. Cryptographic algorithms and identifiers for use with this
specification are described in the separate JSON Web Algorithms (JWA)
specification and IANA registries established by that specification.
https://tools.ietf.org/html/rfc7517
:param token:
:param key:
:param algorithms:
:return:
"""
return jwt.decode(token, key, algorithms=algorithms,
audience=self.client_id, **kwargs)
def load_authorization_config(self, path):
"""
Load Keycloak settings (authorization)
:param path: settings file (json)
:return:
"""
authorization_file = open(path, 'r')
authorization_json = json.loads(authorization_file.read())
self.authorization.load_config(authorization_json)
authorization_file.close()
def get_policies(self, token, method_token_info='introspect', **kwargs):
"""
Get policies by user token
:param token: user token
:return: policies list
"""
if not self.authorization.policies:
raise KeycloakAuthorizationConfigError(
"Keycloak settings not found. Load Authorization Keycloak settings."
)
token_info = self._token_info(token, method_token_info, **kwargs)
if method_token_info == 'introspect' and not token_info['active']:
raise KeycloakInvalidTokenError(
"Token expired or invalid."
)
user_resources = token_info['resource_access'].get(self.client_id)
if not user_resources:
return None
policies = []
for policy_name, policy in self.authorization.policies.items():
for role in user_resources['roles']:
if self._build_name_role(role) in policy.roles:
policies.append(policy)
return list(set(policies))
def get_permissions(self, token, method_token_info='introspect', **kwargs):
"""
Get permission by user token
:param token: user token
:param method_token_info: Decode token method
:param kwargs: parameters for decode
:return: permissions list
"""
if not self.authorization.policies:
raise KeycloakAuthorizationConfigError(
"Keycloak settings not found. Load Authorization Keycloak settings."
)
token_info = self._token_info(token, method_token_info, **kwargs)
if method_token_info == 'introspect' and not token_info['active']:
raise KeycloakInvalidTokenError(
"Token expired or invalid."
)
user_resources = token_info['resource_access'].get(self.client_id)
if not user_resources:
return None
permissions = []
for policy_name, policy in self.authorization.policies.items():
for role in user_resources['roles']:
if self._build_name_role(role) in policy.roles:
permissions += policy.permissions
return list(set(permissions))
from .keycloak_openid import *
from .keycloak_admin import *

19
keycloak/connection.py

@ -162,3 +162,22 @@ class ConnectionManager(object):
except Exception as e:
raise KeycloakConnectionError(
"Can't connect to server (%s)" % e)
def raw_delete(self, path, **kwargs):
""" Submit delete request to the path.
:arg
path (str): Path for request.
:return
Response the request.
:exception
HttpError: Can't connect to server.
"""
try:
return requests.delete(urljoin(self.base_url, path),
params=kwargs,
headers=self.headers,
timeout=self.timeout)
except Exception as e:
raise KeycloakConnectionError(
"Can't connect to server (%s)" % e)

304
keycloak/keycloak_admin.py

@ -0,0 +1,304 @@
# -*- coding: utf-8 -*-
#
# 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.
#
# 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.
#
# 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/>.
from .urls_patterns import URL_ADMIN_USERS_COUNT, URL_ADMIN_USER, URL_ADMIN_USER_CONSENTS, \
URL_ADMIN_SEND_UPDATE_ACCOUNT, URL_ADMIN_RESET_PASSWORD, URL_ADMIN_SEND_VERIFY_EMAIL, URL_ADMIN_GET_SESSIONS, \
URL_ADMIN_SERVER_INFO, URL_ADMIN_CLIENTS, URL_ADMIN_CLIENT, URL_ADMIN_CLIENT_ROLES, URL_ADMIN_REALM_ROLES
from .keycloak_openid import KeycloakOpenID
from .exceptions import raise_error_from_response, KeycloakGetError
from .urls_patterns import (
URL_ADMIN_USERS,
)
from .connection import ConnectionManager
import json
class KeycloakAdmin:
def __init__(self, server_url, username, password, realm_name='master', client_id='admin-cli'):
self._username = username
self._password = password
self._client_id = client_id
self._realm_name = realm_name
# Get token Admin
keycloak_openid = KeycloakOpenID(server_url, client_id, realm_name)
self._token = keycloak_openid.token(username, password)
self._connection = ConnectionManager(base_url=server_url,
headers={'Authorization': 'Bearer ' + self.token.get('access_token'),
'Content-Type': 'application/json'},
timeout=60)
@property
def realm_name(self):
return self._realm_name
@realm_name.setter
def realm_name(self, value):
self._realm_name = value
@property
def connection(self):
return self._connection
@connection.setter
def connection(self, value):
self._connection = value
@property
def client_id(self):
return self._client_id
@client_id.setter
def client_id(self, value):
self._client_id = value
@property
def username(self):
return self._username
@username.setter
def username(self, value):
self._username = value
@property
def password(self):
return self._password
@password.setter
def password(self, value):
self._password = value
@property
def token(self):
return self._token
@token.setter
def token(self, value):
self._token = value
def get_users(self, query=None):
"""
Get users Returns a list of users, filtered according to query parameters
:return: users list
"""
params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_ADMIN_USERS.format(**params_path), **query)
return raise_error_from_response(data_raw, KeycloakGetError)
def create_user(self, payload):
"""
Create a new user Username must be unique
UserRepresentation
http://www.keycloak.org/docs-api/3.3/rest-api/index.html#_userrepresentation
:param payload: UserRepresentation
:return: UserRepresentation
"""
params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_post(URL_ADMIN_USERS.format(**params_path),
data=json.dumps(payload))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=201)
def users_count(self):
"""
User counter
:return: counter
"""
params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_ADMIN_USERS_COUNT.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def get_user(self, user_id):
"""
Get representation of the user
:param user_id: User id
UserRepresentation: http://www.keycloak.org/docs-api/3.3/rest-api/index.html#_userrepresentation
:return: UserRepresentation
"""
params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.connection.raw_get(URL_ADMIN_USER.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def update_user(self, user_id, payload):
"""
Update the user
:param user_id: User id
:param payload: UserRepresentation
:return: Http response
"""
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))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def delete_user(self, user_id):
"""
Delete the user
:param user_id: User id
:return: Http response
"""
params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.connection.raw_delete(URL_ADMIN_USER.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def consents_user(self, user_id):
"""
Get consents granted by the user
:param user_id: User id
:return: consents
"""
params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.connection.raw_get(URL_ADMIN_USER_CONSENTS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def send_update_account(self, user_id, payload, client_id=None, lifespan=None, redirect_uri=None):
"""
Send a update account email to the user An email contains a
link the user can click to perform a set of required actions.
:param user_id:
:param payload:
:param client_id:
:param lifespan:
:param redirect_uri:
:return:
"""
params_path = {"realm-name": self.realm_name, "id": user_id}
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)
return raise_error_from_response(data_raw, KeycloakGetError)
def send_verify_email(self, user_id, client_id=None, redirect_uri=None):
"""
Send a update account email to the user An email contains a
link the user can click to perform a set of required actions.
:param user_id: User id
:param client_id: Client id
:param redirect_uri: Redirect uri
:return:
"""
params_path = {"realm-name": self.realm_name, "id": user_id}
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)
return raise_error_from_response(data_raw, KeycloakGetError)
def get_sessions(self, user_id):
"""
Get sessions associated with the user
:param user_id: User id
UserSessionRepresentation
http://www.keycloak.org/docs-api/3.3/rest-api/index.html#_usersessionrepresentation
:return: UserSessionRepresentation
"""
params_path = {"realm-name": self.realm_name, "id": user_id}
data_raw = self.connection.raw_get(URL_ADMIN_GET_SESSIONS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def get_server_info(self):
"""
Get themes, social providers, auth providers, and event listeners available on this server
ServerInfoRepresentation
http://www.keycloak.org/docs-api/3.3/rest-api/index.html#_serverinforepresentation
:return: ServerInfoRepresentation
"""
data_raw = self.connection.raw_get(URL_ADMIN_SERVER_INFO)
return raise_error_from_response(data_raw, KeycloakGetError)
def get_clients(self):
"""
Get clients belonging to the realm Returns a list of clients belonging to the realm
ClientRepresentation
http://www.keycloak.org/docs-api/3.3/rest-api/index.html#_clientrepresentation
:return: ClientRepresentation
"""
params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_ADMIN_CLIENTS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def get_client(self, client_id):
"""
Get representation of the client
ClientRepresentation
http://www.keycloak.org/docs-api/3.3/rest-api/index.html#_clientrepresentation
:param client_id: id of client (not client-id)
:return: ClientRepresentation
"""
params_path = {"realm-name": self.realm_name, "id": client_id}
data_raw = self.connection.raw_get(URL_ADMIN_CLIENT.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def get_client_role(self, client_id):
"""
Get all roles for the client
RoleRepresentation
http://www.keycloak.org/docs-api/3.3/rest-api/index.html#_rolerepresentation
:param client_id: id of client (not client-id)
:return: RoleRepresentation
"""
params_path = {"realm-name": self.realm_name, "id": client_id}
data_raw = self.connection.raw_get(URL_ADMIN_CLIENT_ROLES.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def get_roles(self):
"""
Get all roles for the realm or client
RoleRepresentation
http://www.keycloak.org/docs-api/3.3/rest-api/index.html#_rolerepresentation
:return: RoleRepresentation
"""
params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_ADMIN_REALM_ROLES.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)

365
keycloak/keycloak_openid.py

@ -0,0 +1,365 @@
# -*- coding: utf-8 -*-
#
# 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.
#
# 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.
#
# 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/>.
from .authorization import Authorization
from .exceptions import raise_error_from_response, KeycloakGetError, \
KeycloakRPTNotFound, KeycloakAuthorizationConfigError, KeycloakInvalidTokenError
from .urls_patterns import (
URL_TOKEN,
URL_USERINFO,
URL_WELL_KNOWN,
URL_LOGOUT,
URL_CERTS,
URL_ENTITLEMENT,
URL_INTROSPECT
)
from .connection import ConnectionManager
from jose import jwt
import json
class KeycloakOpenID:
def __init__(self, server_url, client_id, realm_name, client_secret_key=None):
self._client_id = client_id
self._client_secret_key = client_secret_key
self._realm_name = realm_name
self._connection = ConnectionManager(base_url=server_url,
headers={},
timeout=60)
self._authorization = Authorization()
@property
def client_id(self):
return self._client_id
@client_id.setter
def client_id(self, 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 realm_name(self):
return self._realm_name
@realm_name.setter
def realm_name(self, value):
self._realm_name = value
@property
def connection(self):
return self._connection
@connection.setter
def connection(self, value):
self._connection = value
@property
def authorization(self):
return self._authorization
@authorization.setter
def authorization(self, value):
self._authorization = value
def _add_secret_key(self, payload):
"""
Add secret key if exist.
:param payload:
:return:
"""
if self.client_secret_key:
payload.update({"client_secret": self.client_secret_key})
return payload
def _build_name_role(self, role):
"""
:param role:
:return:
"""
return self.client_id + "/" + role
def _token_info(self, token, method_token_info, **kwargs):
"""
:param token:
:param method_token_info:
:param kwargs:
:return:
"""
if method_token_info == 'introspect':
token_info = self.introspect(token)
else:
token_info = self.decode_token(token, **kwargs)
return token_info
def well_know(self):
""" The most important endpoint to understand is the well-known configuration
endpoint. It lists endpoints and other configuration options relevant to
the OpenID Connect implementation in Keycloak.
:return It lists endpoints and other configuration options relevant.
"""
params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_WELL_KNOWN.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def auth_url(self, redirect_uri):
"""
http://openid.net/specs/openid-connect-core-1_0.html#AuthorizationEndpoint
:return:
"""
return NotImplemented
def token(self, username, password, grant_type=["password"]):
"""
The token endpoint is used to obtain tokens. Tokens can either be obtained by
exchanging an authorization code or by supplying credentials directly depending on
what flow is used. The token endpoint is also used to obtain new access tokens
when they expire.
http://openid.net/specs/openid-connect-core-1_0.html#TokenEndpoint
:param username:
:param password:
:param grant_type:
:return:
"""
params_path = {"realm-name": self.realm_name}
payload = {"username": username, "password": password,
"client_id": self.client_id, "grant_type": grant_type}
payload = self._add_secret_key(payload)
data_raw = self.connection.raw_post(URL_TOKEN.format(**params_path),
data=payload)
return raise_error_from_response(data_raw, KeycloakGetError)
def userinfo(self, token):
"""
The userinfo endpoint returns standard claims about the authenticated user,
and is protected by a bearer token.
http://openid.net/specs/openid-connect-core-1_0.html#UserInfo
:param token:
:return:
"""
self.connection.add_param_headers("Authorization", "Bearer " + token)
params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_USERINFO.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def logout(self, refresh_token):
"""
The logout endpoint logs out the authenticated user.
:param refresh_token:
:return:
"""
params_path = {"realm-name": self.realm_name}
payload = {"client_id": self.client_id, "refresh_token": refresh_token}
payload = self._add_secret_key(payload)
data_raw = self.connection.raw_post(URL_LOGOUT.format(**params_path),
data=payload)
return raise_error_from_response(data_raw, KeycloakGetError, expected_code=204)
def certs(self):
"""
The certificate endpoint returns the public keys enabled by the realm, encoded as a
JSON Web Key (JWK). Depending on the realm settings there can be one or more keys enabled
for verifying tokens.
https://tools.ietf.org/html/rfc7517
:return:
"""
params_path = {"realm-name": self.realm_name}
data_raw = self.connection.raw_get(URL_CERTS.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def entitlement(self, token, resource_server_id):
"""
Client applications can use a specific endpoint to obtain a special security token
called a requesting party token (RPT). This token consists of all the entitlements
(or permissions) for a user as a result of the evaluation of the permissions and authorization
policies associated with the resources being requested. With an RPT, client applications can
gain access to protected resources at the resource server.
:return:
"""
self.connection.add_param_headers("Authorization", "Bearer " + token)
params_path = {"realm-name": self.realm_name, "resource-server-id": resource_server_id}
data_raw = self.connection.raw_get(URL_ENTITLEMENT.format(**params_path))
return raise_error_from_response(data_raw, KeycloakGetError)
def introspect(self, token, rpt=None, token_type_hint=None):
"""
The introspection endpoint is used to retrieve the active state of a token. It is can only be
invoked by confidential clients.
https://tools.ietf.org/html/rfc7662
:param token:
:param rpt:
:param token_type_hint:
:return:
"""
params_path = {"realm-name": self.realm_name}
payload = {"client_id": self.client_id, "token": token}
if token_type_hint == 'requesting_party_token':
if rpt:
payload.update({"token": rpt, "token_type_hint": token_type_hint})
self.connection.add_param_headers("Authorization", "Bearer " + token)
else:
raise KeycloakRPTNotFound("Can't found RPT.")
payload = self._add_secret_key(payload)
data_raw = self.connection.raw_post(URL_INTROSPECT.format(**params_path),
data=payload)
return raise_error_from_response(data_raw, KeycloakGetError)
def decode_token(self, token, key, algorithms=['RS256'], **kwargs):
"""
A JSON Web Key (JWK) is a JavaScript Object Notation (JSON) data
structure that represents a cryptographic key. This specification
also defines a JWK Set JSON data structure that represents a set of
JWKs. Cryptographic algorithms and identifiers for use with this
specification are described in the separate JSON Web Algorithms (JWA)
specification and IANA registries established by that specification.
https://tools.ietf.org/html/rfc7517
:param token:
:param key:
:param algorithms:
:return:
"""
return jwt.decode(token, key, algorithms=algorithms,
audience=self.client_id, **kwargs)
def load_authorization_config(self, path):
"""
Load Keycloak settings (authorization)
:param path: settings file (json)
:return:
"""
authorization_file = open(path, 'r')
authorization_json = json.loads(authorization_file.read())
self.authorization.load_config(authorization_json)
authorization_file.close()
def get_policies(self, token, method_token_info='introspect', **kwargs):
"""
Get policies by user token
:param token: user token
:return: policies list
"""
if not self.authorization.policies:
raise KeycloakAuthorizationConfigError(
"Keycloak settings not found. Load Authorization Keycloak settings."
)
token_info = self._token_info(token, method_token_info, **kwargs)
if method_token_info == 'introspect' and not token_info['active']:
raise KeycloakInvalidTokenError(
"Token expired or invalid."
)
user_resources = token_info['resource_access'].get(self.client_id)
if not user_resources:
return None
policies = []
for policy_name, policy in self.authorization.policies.items():
for role in user_resources['roles']:
if self._build_name_role(role) in policy.roles:
policies.append(policy)
return list(set(policies))
def get_permissions(self, token, method_token_info='introspect', **kwargs):
"""
Get permission by user token
:param token: user token
:param method_token_info: Decode token method
:param kwargs: parameters for decode
:return: permissions list
"""
if not self.authorization.policies:
raise KeycloakAuthorizationConfigError(
"Keycloak settings not found. Load Authorization Keycloak settings."
)
token_info = self._token_info(token, method_token_info, **kwargs)
if method_token_info == 'introspect' and not token_info['active']:
raise KeycloakInvalidTokenError(
"Token expired or invalid."
)
user_resources = token_info['resource_access'].get(self.client_id)
if not user_resources:
return None
permissions = []
for policy_name, policy in self.authorization.policies.items():
for role in user_resources['roles']:
if self._build_name_role(role) in policy.roles:
permissions += policy.permissions
return list(set(permissions))

22
keycloak/urls_patterns.py

@ -15,12 +15,30 @@
# 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/>.
# OPENID URLS
URL_WELL_KNOWN = "realms/{realm-name}/.well-known/openid-configuration"
URL_AUTH = "realms/{realm-name}/protocol/openid-connect/auth"
URL_TOKEN = "realms/{realm-name}/protocol/openid-connect/token"
URL_USERINFO = "realms/{realm-name}/protocol/openid-connect/userinfo"
URL_LOGOUT = "realms/{realm-name}/protocol/openid-connect/logout"
URL_CERTS = "realms/{realm-name}/protocol/openid-connect/certs"
URL_INTROSPECT = "realms/{realm-name}/protocol/openid-connect/token/introspect"
URL_ENTITLEMENT = "realms/{realm-name}/authz/entitlement/{resource-server-id}"
# ADMIN URLS
URL_ADMIN_USERS = "admin/realms/{realm-name}/users"
URL_ADMIN_USERS_COUNT = "admin/realms/{realm-name}/users/count"
URL_ADMIN_USER = "admin/realms/{realm-name}/users/{id}"
URL_ADMIN_USER_CONSENTS = "admin/realms/{realm-name}/users/{id}/consents"
URL_ADMIN_SEND_UPDATE_ACCOUNT = "admin/realms/{realm-name}/users/{id}/execute-actions-email"
URL_ADMIN_SEND_VERIFY_EMAIL = "admin/realms/{realm-name}/users/{id}/send-verify-email"
URL_ADMIN_RESET_PASSWORD = "admin/realms/{realm-name}/users/{id}/reset-password"
URL_ADMIN_GET_SESSIONS = "admin/realms/{realm-name}/users/{id}/sessions"
URL_ADMIN_SERVER_INFO = "admin/serverinfo"
URL_ADMIN_CLIENTS = "admin/realms/{realm-name}/clients"
URL_ADMIN_CLIENT = "admin/realms/{realm-name}/clients/{id}"
URL_ADMIN_CLIENT_ROLES = "admin/realms/{realm-name}/clients/{id}/roles"
URL_ADMIN_REALM_ROLES = "admin/realms/{realm-name}/roles"

2
setup.py

@ -4,7 +4,7 @@ from setuptools import setup
setup(
name='python-keycloak',
version='0.8.4',
version='0.9.0',
url='https://bitbucket.org/agriness/python-keycloak',
license='GNU General Public License - V3',
author='Marcos Pereira',

Loading…
Cancel
Save