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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
|
# 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 abc
import logging
from oslo.config import cfg
import six
from keystoneclient import _discover
from keystoneclient.auth import base
from keystoneclient import exceptions
LOG = logging.getLogger(__name__)
@six.add_metaclass(abc.ABCMeta)
class BaseIdentityPlugin(base.BaseAuthPlugin):
# we count a token as valid if it is valid for at least this many seconds
MIN_TOKEN_LIFE_SECONDS = 1
def __init__(self,
auth_url=None,
username=None,
password=None,
token=None,
trust_id=None):
super(BaseIdentityPlugin, self).__init__()
self.auth_url = auth_url
self.auth_ref = None
self._endpoint_cache = {}
# NOTE(jamielennox): DEPRECATED. The following should not really be set
# here but handled by the individual auth plugin.
self.username = username
self.password = password
self.token = token
self.trust_id = trust_id
@abc.abstractmethod
def get_auth_ref(self, session, **kwargs):
"""Obtain a token from an OpenStack Identity Service.
This method is overridden by the various token version plugins.
This function should not be called independently and is expected to be
invoked via the do_authenticate function.
This function will be invoked if the AcessInfo object cached by the
plugin is not valid. Thus plugins should always fetch a new AccessInfo
when invoked. If you are looking to just retrieve the current auth
data then you should use get_access.
:raises InvalidResponse: The response returned wasn't appropriate.
:raises HttpError: An error from an invalid HTTP response.
:returns AccessInfo: Token access information.
"""
def get_token(self, session, **kwargs):
"""Return a valid auth token.
If a valid token is not present then a new one will be fetched.
:raises HttpError: An error from an invalid HTTP response.
:return string: A valid token.
"""
return self.get_access(session).auth_token
def get_access(self, session, **kwargs):
"""Fetch or return a current AccessInfo object.
If a valid AccessInfo is present then it is returned otherwise a new
one will be fetched.
:raises HttpError: An error from an invalid HTTP response.
:returns AccessInfo: Valid AccessInfo
"""
if (not self.auth_ref or
self.auth_ref.will_expire_soon(self.MIN_TOKEN_LIFE_SECONDS)):
self.auth_ref = self.get_auth_ref(session)
return self.auth_ref
def invalidate(self):
"""Invalidate the current authentication data.
This should result in fetching a new token on next call.
A plugin may be invalidated if an Unauthorized HTTP response is
returned to indicate that the token may have been revoked or is
otherwise now invalid.
:returns bool: True if there was something that the plugin did to
invalidate. This means that it makes sense to try again.
If nothing happens returns False to indicate give up.
"""
self.auth_ref = None
return True
def get_endpoint(self, session, service_type=None, interface=None,
region_name=None, service_name=None, version=None,
**kwargs):
"""Return a valid endpoint for a service.
If a valid token is not present then a new one will be fetched using
the session and kwargs.
:param string service_type: The type of service to lookup the endpoint
for. This plugin will return None (failure)
if service_type is not provided.
:param string interface: The exposure of the endpoint. Should be
`public`, `internal` or `admin`.
Defaults to `public`.
:param string region_name: The region the endpoint should exist in.
(optional)
:param string service_name: The name of the service in the catalog.
(optional)
:param tuple version: The minimum version number required for this
endpoint. (optional)
:raises HttpError: An error from an invalid HTTP response.
:return string or None: A valid endpoint URL or None if not available.
"""
if not service_type:
LOG.warn('Plugin cannot return an endpoint without knowing the '
'service type that is required. Add service_type to '
'endpoint filtering data.')
return None
if not interface:
interface = 'public'
service_catalog = self.get_access(session).service_catalog
sc_url = service_catalog.url_for(service_type=service_type,
endpoint_type=interface,
region_name=region_name,
service_name=service_name)
if not version:
# NOTE(jamielennox): This may not be the best thing to default to
# but is here for backwards compatibility. It may be worth
# defaulting to the most recent version.
return sc_url
disc = None
# NOTE(jamielennox): we want to cache endpoints on the session as well
# so that they maintain sharing between auth plugins. Create a cache on
# the session if it doesn't exist already.
try:
session_endpoint_cache = session._identity_endpoint_cache
except AttributeError:
session_endpoint_cache = session._identity_endpoint_cache = {}
# NOTE(jamielennox): There is a cache located on both the session
# object and the auth plugin object so that they can be shared and the
# cache is still usable
for cache in (self._endpoint_cache, session_endpoint_cache):
disc = cache.get(sc_url)
if disc:
break
else:
try:
disc = _discover.Discover(session, sc_url)
except (exceptions.HTTPError, exceptions.ConnectionError):
LOG.warn('Failed to contact the endpoint at %s for discovery. '
'Fallback to using that endpoint as the '
'base url.', sc_url)
return sc_url
else:
self._endpoint_cache[sc_url] = disc
session_endpoint_cache[sc_url] = disc
return disc.url_for(version)
@classmethod
def get_options(cls):
options = super(BaseIdentityPlugin, cls).get_options()
options.extend([
cfg.StrOpt('auth-url', help='Authentication URL'),
])
return options
|
