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
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
|
# 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.
#
"""OpenStackConfig subclass for argument compatibility"""
import logging
import typing as ty
from keystoneauth1.loading import identity as ksa_loading
from openstack.config import exceptions as sdk_exceptions
from openstack.config import loader as config
from oslo_utils import strutils
LOG = logging.getLogger(__name__)
# Subclass OpenStackConfig in order to munge config values
# before auth plugins are loaded
class OSC_Config(config.OpenStackConfig): # type: ignore
def _auth_select_default_plugin(
self, config: dict[str, ty.Any]
) -> dict[str, ty.Any]:
"""Select a default plugin based on supplied arguments
Migrated from auth.select_auth_plugin()
"""
identity_version = str(config.get('identity_api_version', ''))
if config.get('username', None) and not config.get('auth_type', None):
if identity_version == '3':
config['auth_type'] = 'v3password'
elif identity_version.startswith('2'):
config['auth_type'] = 'v2password'
else:
# let keystoneauth figure it out itself
config['auth_type'] = 'password'
elif config.get('token', None) and not config.get('auth_type', None):
if identity_version == '3':
config['auth_type'] = 'v3token'
elif identity_version.startswith('2'):
config['auth_type'] = 'v2token'
else:
# let keystoneauth figure it out itself
config['auth_type'] = 'token'
else:
# The ultimate default is similar to the original behaviour,
# but this time with version discovery
if not config.get('auth_type', None):
config['auth_type'] = 'password'
LOG.debug("Auth plugin {} selected".format(config['auth_type']))
return config
def _auth_v2_arguments(
self, config: dict[str, ty.Any]
) -> dict[str, ty.Any]:
"""Set up v2-required arguments from v3 info
Migrated from auth.build_auth_params()
"""
if 'auth_type' in config and config['auth_type'].startswith("v2"):
if 'project_id' in config['auth']:
config['auth']['tenant_id'] = config['auth']['project_id']
if 'project_name' in config['auth']:
config['auth']['tenant_name'] = config['auth']['project_name']
return config
def _auth_v2_ignore_v3(
self, config: dict[str, ty.Any]
) -> dict[str, ty.Any]:
"""Remove v3 arguments if present for v2 plugin
Migrated from clientmanager.setup_auth()
"""
# NOTE(hieulq): If USER_DOMAIN_NAME, USER_DOMAIN_ID, PROJECT_DOMAIN_ID
# or PROJECT_DOMAIN_NAME is present and API_VERSION is 2.0, then
# ignore all domain related configs.
if not str(config.get('identity_api_version', '')).startswith('2'):
return config
if not config.get('auth_type') or not config['auth_type'].endswith(
'password'
):
return config
domain_props = [
'project_domain_id',
'project_domain_name',
'user_domain_id',
'user_domain_name',
]
for prop in domain_props:
if config['auth'].pop(prop, None) is not None:
if config.get('cloud'):
LOG.warning(
"Ignoring domain related config {} for {}"
"because identity API version is 2.0".format(
prop, config['cloud']
)
)
else:
LOG.warning(
f"Ignoring domain related config {prop} because"
" identity API version is 2.0"
)
return config
def _auth_default_domain(
self, config: dict[str, ty.Any]
) -> dict[str, ty.Any]:
"""Set a default domain from available arguments
Migrated from clientmanager.setup_auth()
"""
identity_version = str(config.get('identity_api_version', ''))
auth_type = config.get('auth_type', None)
# TODO(mordred): This is a usability improvement that's broadly useful
# We should port it back up into os-client-config.
default_domain = config.get('default_domain', None)
if (
identity_version == '3'
and (auth_type and not auth_type.startswith('v2'))
and default_domain
):
# NOTE(stevemar): If PROJECT_DOMAIN_ID or PROJECT_DOMAIN_NAME is
# present, then do not change the behaviour. Otherwise, set the
# PROJECT_DOMAIN_ID to 'OS_DEFAULT_DOMAIN' for better usability.
if (
auth_type in ("password", "v3password", "v3totp")
and not config['auth'].get('project_domain_id')
and not config['auth'].get('project_domain_name')
):
config['auth']['project_domain_id'] = default_domain
# NOTE(stevemar): If USER_DOMAIN_ID or USER_DOMAIN_NAME is present,
# then do not change the behaviour. Otherwise, set the
# USER_DOMAIN_ID to 'OS_DEFAULT_DOMAIN' for better usability.
# NOTE(aloga): this should only be set if there is a username.
# TODO(dtroyer): Move this to os-client-config after the plugin has
# been loaded so we can check directly if the options are accepted.
if (
auth_type in ("password", "v3password", "v3totp")
and not config['auth'].get('user_domain_id')
and not config['auth'].get('user_domain_name')
):
config['auth']['user_domain_id'] = default_domain
return config
def auth_config_hook(self, config: dict[str, ty.Any]) -> dict[str, ty.Any]:
"""Allow examination of config values before loading auth plugin
OpenStackClient will override this to perform additional checks
on auth_type.
"""
config = self._auth_select_default_plugin(config)
config = self._auth_v2_arguments(config)
config = self._auth_v2_ignore_v3(config)
config = self._auth_default_domain(config)
if LOG.isEnabledFor(logging.DEBUG):
LOG.debug(
"auth_config_hook(): %s", strutils.mask_password(str(config))
)
return config
def _validate_auth(
self,
config: dict[str, ty.Any],
loader: ksa_loading.BaseIdentityLoader[ty.Any],
) -> dict[str, ty.Any]:
"""Validate auth plugin arguments"""
# May throw a keystoneauth1.exceptions.NoMatchingPlugin
plugin_options = loader.get_options()
msgs = []
prompt_options = []
for p_opt in plugin_options:
# if it's in config, win, move it and kill it from config dict
# if it's in config.auth but not in config we're good
# deprecated loses to current
# provided beats default, deprecated or not
winning_value = self._find_winning_auth_value(p_opt, config)
if not winning_value:
winning_value = self._find_winning_auth_value(
p_opt, config['auth']
)
# if the plugin tells us that this value is required
# then error if it's doesn't exist now
if not winning_value and p_opt.required:
msgs.append(
'Missing value {auth_key}'
' required for auth plugin {plugin}'.format(
auth_key=p_opt.name,
plugin=config.get('auth_type'),
)
)
# Clean up after ourselves
for opt in [p_opt.name] + [o.name for o in p_opt.deprecated]:
opt = opt.replace('-', '_')
config.pop(opt, None)
config['auth'].pop(opt, None)
if winning_value:
# Prefer the plugin configuration dest value if the value's key
# is marked as depreciated.
if p_opt.dest is None:
config['auth'][p_opt.name.replace('-', '_')] = (
winning_value
)
else:
config['auth'][p_opt.dest] = winning_value
# See if this needs a prompting
if (
'prompt' in vars(p_opt)
and p_opt.prompt is not None
and p_opt.dest not in config['auth']
and self._pw_callback is not None
):
# Defer these until we know all required opts are present
prompt_options.append(p_opt)
if msgs:
raise sdk_exceptions.OpenStackConfigException('\n'.join(msgs))
else:
for p_opt in prompt_options:
config['auth'][p_opt.dest] = self._pw_callback(p_opt.prompt)
return config
# TODO(stephenfin): Add type once we have typing for SDK
def load_auth_plugin(self, config: dict[str, ty.Any]) -> ty.Any:
"""Get auth plugin and validate args"""
loader = self._get_auth_loader(config)
config = self._validate_auth(config, loader)
auth_plugin = loader.load_from_options(**config['auth'])
return auth_plugin
|
