File: keycloak.py

package info (click to toggle)
python-mistralclient 1%3A2.1.2-1
  • links: PTS, VCS
  • area: main
  • in suites: stretch
  • size: 888 kB
  • ctags: 1,301
  • sloc: python: 8,941; sh: 216; makefile: 26
file content (141 lines) | stat: -rw-r--r-- 4,321 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
# Copyright 2016 - Nokia Networks
#
#    Licensed under the Apache License, Version 2.0 (the "License");
#    you may not use this file except in compliance with the License.
#    You may obtain a copy of the License at
#
#        http://www.apache.org/licenses/LICENSE-2.0
#
#    Unless required by applicable law or agreed to in writing, software
#    distributed under the License is distributed on an "AS IS" BASIS,
#    WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
#    See the License for the specific language governing permissions and
#    limitations under the License.

import logging
import pprint
import requests


LOG = logging.getLogger(__name__)


def authenticate(auth_url, client_id, client_secret, realm_name,
                 username=None, password=None, access_token=None,
                 cacert=None, insecure=False):
    """Performs authentication using Keycloak OpenID Protocol.

    :param auth_url: Base authentication url of KeyCloak server (e.g.
        "https://my.keycloak:8443/auth"
    :param client_id: Client ID (according to OpenID Connect protocol).
    :param client_secret: Client secret (according to OpenID Connect protocol).
    :param realm_name: KeyCloak realm name.
    :param username: User name (Optional, if None then access_token must be
        provided).
    :param password: Password (Optional).
    :param access_token: Access token. If passed, username and password are
        not used and this method just validates the token and refreshes it,
        if needed. (Optional, if None then username must be provided)
    :param cacert: SSL certificate file (Optional).
    :param insecure: If True, SSL certificate is not verified (Optional).

    """
    if not auth_url:
        raise ValueError('Base authentication url is not provided.')

    if not client_id:
        raise ValueError('Client ID is not provided.')

    if not client_secret:
        raise ValueError('Client secret is not provided.')

    if not realm_name:
        raise ValueError('Project(realm) name is not provided.')

    if username and access_token:
        raise ValueError(
            "User name and access token can't be provided at the same time."
        )

    if access_token:
        return _authenticate_with_token(
            auth_url,
            client_id,
            client_secret,
            access_token,
            cacert,
            insecure
        )

    if not username:
        raise ValueError('Either user name or access token must be provided.')

    return _authenticate_with_password(
        auth_url,
        client_id,
        client_secret,
        realm_name,
        username,
        password,
        cacert,
        insecure
    )


def _authenticate_with_token(auth_url, client_id, client_secret, auth_token,
                             cacert=None, insecure=None):
    # TODO(rakhmerov): Implement.
    raise NotImplementedError


def _authenticate_with_password(auth_url, client_id, client_secret,
                                realm_name, username, password,
                                cacert=None, insecure=None):
    access_token_endpoint = (
        "%s/realms/%s/protocol/openid-connect/token" % (auth_url, realm_name)
    )

    client_auth = (client_id, client_secret)

    body = {
        'grant_type': 'password',
        'username': username,
        'password': password,
        'scope': 'profile'
    }

    resp = requests.post(
        access_token_endpoint,
        auth=client_auth,
        data=body,
        verify=not insecure
    )

    try:
        resp.raise_for_status()
    except Exception as e:
        raise Exception("Failed to get access token:\n %s" % str(e))

    LOG.debug(
        "HTTP response from OIDC provider: %s" % pprint.pformat(resp.json())
    )

    return resp.json()['access_token']


# An example of using KeyCloak OpenID authentication.

if __name__ == '__main__':
    print("Using username/password to get access token from KeyCloak...")

    a_token = authenticate(
        "https://my.keycloak:8443/auth",
        client_id="mistral_client",
        client_secret="4a080907-921b-409a-b793-c431609c3a47",
        realm_name="mistral",
        username="user",
        password="secret",
        insecure=True
    )

    print("Access token: %s" % a_token)