# Copyright (c) 2017, Eike Frost # # This code is part of Ansible, but is an independent component. # This particular file snippet, and this file snippet only, is BSD licensed. # Modules you write using this snippet, which is embedded dynamically by Ansible # still belong to the author of the module, and may assign their own license # to the complete work. # # Redistribution and use in source and binary forms, with or without modification, # are permitted provided that the following conditions are met: # # * Redistributions of source code must retain the above copyright # notice, this list of conditions and the following disclaimer. # * Redistributions in binary form must reproduce the above copyright notice, # this list of conditions and the following disclaimer in the documentation # and/or other materials provided with the distribution. # # THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND # ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED # WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE DISCLAIMED. # IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, # INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, # PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS # INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT # LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE # USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. from __future__ import absolute_import, division, print_function __metaclass__ = type import json from ansible.module_utils.urls import open_url from ansible.module_utils.six.moves.urllib.parse import urlencode from ansible.module_utils.six.moves.urllib.error import HTTPError URL_TOKEN = "{url}/realms/{realm}/protocol/openid-connect/token" URL_CLIENT = "{url}/admin/realms/{realm}/clients/{id}" URL_CLIENTS = "{url}/admin/realms/{realm}/clients" URL_CLIENT_ROLES = "{url}/admin/realms/{realm}/clients/{id}/roles" URL_REALM_ROLES = "{url}/admin/realms/{realm}/roles" def keycloak_argument_spec(): """ Returns argument_spec of options common to keycloak_*-modules :return: argument_spec dict """ return dict( auth_keycloak_url=dict(type='str', aliases=['url'], required=True), auth_client_id=dict(type='str', default='admin-cli'), auth_realm=dict(type='str', required=True), auth_client_secret=dict(type='str', default=None), auth_username=dict(type='str', aliases=['username'], required=True), auth_password=dict(type='str', aliases=['password'], required=True, no_log=True), validate_certs=dict(type='bool', default=True) ) def camel(words): return words.split('_')[0] + ''.join(x.capitalize() or '_' for x in words.split('_')[1:]) class KeycloakAPI(object): """ Keycloak API access; Keycloak uses OAuth 2.0 to protect its API, an access token for which is obtained through OpenID connect """ def __init__(self, module): self.module = module self.token = None self._connect() def _connect(self): """ Obtains an access_token and saves it for use in API accesses """ self.baseurl = self.module.params.get('auth_keycloak_url') self.validate_certs = self.module.params.get('validate_certs') auth_url = URL_TOKEN.format(url=self.baseurl, realm=self.module.params.get('auth_realm')) payload = {'grant_type': 'password', 'client_id': self.module.params.get('auth_client_id'), 'client_secret': self.module.params.get('auth_client_secret'), 'username': self.module.params.get('auth_username'), 'password': self.module.params.get('auth_password')} # Remove empty items, for instance missing client_secret payload = dict((k, v) for k, v in payload.items() if v is not None) try: r = json.load(open_url(auth_url, method='POST', validate_certs=self.validate_certs, data=urlencode(payload))) except Exception as e: self.module.fail_json(msg='Could not obtain access token from %s: %s' % (auth_url, str(e))) if 'access_token' in r: self.token = r['access_token'] self.restheaders = {'Authorization': 'Bearer ' + self.token, 'Content-Type': 'application/json'} else: self.module.fail_json(msg='Could not obtain access token from %s' % auth_url) def get_clients(self, realm='master', filter=None): """ Obtains client representations for clients in a realm :param realm: realm to be queried :param filter: if defined, only the client with clientId specified in the filter is returned :return: list of dicts of client representations """ clientlist_url = URL_CLIENTS.format(url=self.baseurl, realm=realm) if filter is not None: clientlist_url += '?clientId=%s' % filter try: return json.load(open_url(clientlist_url, method='GET', headers=self.restheaders, validate_certs=self.validate_certs)) except Exception as e: self.module.fail_json(msg='Could not obtain list of clients for realm %s: %s' % (realm, str(e))) def get_client_by_clientid(self, client_id, realm='master'): """ Get client representation by clientId :param client_id: The clientId to be queried :param realm: realm from which to obtain the client representation :return: dict with a client representation or None if none matching exist """ r = self.get_clients(realm=realm, filter=client_id) if len(r) > 0: return r[0] else: return None def get_client_by_id(self, id, realm='master'): """ Obtain client representatio by id :param id: id (not clientId) of client to be queried :param realm: client from this realm :return: dict of client representation or None if none matching exist """ client_url = URL_CLIENT.format(url=self.baseurl, realm=realm, id=id) try: return json.load(open_url(client_url, method='GET', headers=self.restheaders, validate_certs=self.validate_certs)) except HTTPError as e: if e.code == 404: return None else: self.module.fail_json(msg='Could not obtain client %s for realm %s: %s' % (id, realm, str(e))) except Exception as e: self.module.fail_json(msg='Could not obtain client %s for realm %s: %s' % (id, realm, str(e))) def update_client(self, id, clientrep, realm="master"): """ Update an existing client :param id: id (not clientId) of client to be updated in Keycloak :param clientrep: corresponding (partial/full) client representation with updates :param realm: realm the client is in :return: HTTPResponse object on success """ client_url = URL_CLIENT.format(url=self.baseurl, realm=realm, id=id) try: return open_url(client_url, method='PUT', headers=self.restheaders, data=json.dumps(clientrep), validate_certs=self.validate_certs) except Exception as e: self.module.fail_json(msg='Could not update client %s in realm %s: %s' % (id, realm, str(e))) def create_client(self, clientrep, realm="master"): """ Create a client in keycloak :param clientrep: Client representation of client to be created. Must at least contain field clientId :param realm: realm for client to be created :return: HTTPResponse object on success """ client_url = URL_CLIENTS.format(url=self.baseurl, realm=realm) try: return open_url(client_url, method='POST', headers=self.restheaders, data=json.dumps(clientrep), validate_certs=self.validate_certs) except Exception as e: self.module.fail_json(msg='Could not create client %s in realm %s: %s' % (clientrep['clientId'], realm, str(e))) def delete_client(self, id, realm="master"): """ Delete a client from Keycloak :param id: id (not clientId) of client to be deleted :param realm: realm of client to be deleted :return: HTTPResponse object on success """ client_url = URL_CLIENT.format(url=self.baseurl, realm=realm, id=id) try: return open_url(client_url, method='DELETE', headers=self.restheaders, validate_certs=self.validate_certs) except Exception as e: self.module.fail_json(msg='Could not delete client %s in realm %s: %s' % (id, realm, str(e)))