Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
if not ctxt:
msg = _("User is not authorized to use key manager.")
LOG.error(msg)
raise exception.Forbidden(msg)
if not hasattr(ctxt, 'project_id') or ctxt.project_id is None:
msg = _("Unable to create Barbican Client without project_id.")
LOG.error(msg)
raise exception.KeyManagerError(msg)
# If same context, return cached barbican client
if self._barbican_client and self._current_context == ctxt:
return self._barbican_client
try:
_SESSION = ks_loading.load_session_from_conf_options(
CONF,
nova.conf.barbican.barbican_group)
auth = ctxt.get_auth_plugin()
service_type, service_name, interface = (CONF.
barbican.
catalog_info.
split(':'))
region_name = CONF.barbican.os_region_name
service_parameters = {'service_type': service_type,
'service_name': service_name,
'interface': interface,
'region_name': region_name}
if CONF.barbican.endpoint_template:
self._base_url = (CONF.barbican.endpoint_template %
def __init__(self, **kwargs):
super(GnocchiStorage, self).__init__(**kwargs)
conf = kwargs.get('conf') or ck_utils.load_conf(
CONF.collect.metrics_conf)
self.conf = validate_conf(conf)
self.auth = ks_loading.load_auth_from_conf_options(
CONF,
GNOCCHI_STORAGE_OPTS)
self.session = ks_loading.load_session_from_conf_options(
CONF,
GNOCCHI_STORAGE_OPTS,
auth=self.auth)
self._conn = gclient.Client(
'1',
session=self.session,
adapter_options={'connect_retries': 3,
'interface': CONF.storage_gnocchi.interface})
self._archive_policy_name = (
CONF.storage_gnocchi.archive_policy_name)
self._archive_policy_definition = json.loads(
CONF.storage_gnocchi.archive_policy_definition)
self._period = kwargs.get('period') or CONF.collect.period
self._measurements = dict()
self._resource_type_data = dict()
self._init_resource_types()
def _get_session(self, group):
auth = ks_loading.load_auth_from_conf_options(cfg.CONF, group)
session = ks_loading.load_session_from_conf_options(
cfg.CONF, group, auth=auth)
return session
def _get_client(self, retry_on_conflict=True):
max_retries = CONF.ironic.api_max_retries if retry_on_conflict else 1
retry_interval = (CONF.ironic.api_retry_interval
if retry_on_conflict else 0)
# If we've already constructed a valid, authed client, just return
# that.
if retry_on_conflict and self._cached_client is not None:
return self._cached_client
auth_plugin = self._get_auth_plugin()
sess = ks_loading.load_session_from_conf_options(CONF,
IRONIC_GROUP.name,
auth=auth_plugin)
# Retries for Conflict exception
kwargs = {}
kwargs['max_retries'] = max_retries
kwargs['retry_interval'] = retry_interval
# NOTE(TheJulia): The ability for a list of available versions to be
# accepted was added in python-ironicclient 2.2.0. The highest
# available version will be utilized by the client for the lifetime
# of the client.
kwargs['os_ironic_api_version'] = [
'%d.%d' % IRONIC_API_VERSION, '%d.%d' % PRIOR_IRONIC_API_VERSION]
ironic_conf = CONF[IRONIC_GROUP.name]
# valid_interfaces is a list. ironicclient passes this kwarg through to
def _get_keystone_session(self):
auth = ka_loading.load_auth_from_conf_options(CONF,
_CLIENTS_AUTH_GROUP)
sess = ka_loading.load_session_from_conf_options(CONF,
_CLIENTS_AUTH_GROUP,
auth=auth)
return sess
def _load_session():
global _SESSION
if not _SESSION:
_SESSION = ks_loading.load_session_from_conf_options(
CONF, nova.conf.cinder.cinder_group.name)
def _build_trust(self):
# Get zaqar trustee user ID for trust building
auth = ks_loading.load_auth_from_conf_options(CONF, 'zaqar')
session = ks_loading.load_session_from_conf_options(CONF, 'zaqar')
zaqar_trustee_user_id = session.get_user_id(auth=auth)
try:
trust = self.keystone().trust_get_by_trustor(self.user,
zaqar_trustee_user_id,
self.project)
if not trust:
# Create a trust if no existing one found
roles = self.notifier_roles
for role in roles:
# Remove 'admin' role from delegated roles list
# unless it is the only role user has
if role == 'admin' and len(roles) > 1:
roles.remove(role)
trust = self.keystone().trust_create(self.user,
zaqar_trustee_user_id,
self.project,
def __init__(self, **kwargs):
super(GnocchiStorage, self).__init__(**kwargs)
self.auth = ks_loading.load_auth_from_conf_options(
CONF,
GNOCCHI_STORAGE_OPTS)
self.session = ks_loading.load_session_from_conf_options(
CONF,
GNOCCHI_STORAGE_OPTS,
auth=self.auth)
self._conn = gclient.Client(
'1',
session=self.session,
adapter_options={'connect_retries': 3,
'interface': CONF.storage_gnocchi.interface})
self._measures = {}
self._archive_policy_name = (
CONF.storage_gnocchi.archive_policy_name)
self._archive_policy_definition = json.loads(
CONF.storage_gnocchi.archive_policy_definition)
self._period = METRICS_CONF['period']
if "period" in kwargs:
self._period = kwargs["period"]
what's configured.
"""
global _ADMIN_AUTH
global _SESSION
if not _ADMIN_AUTH:
_ADMIN_AUTH = ks_loading.load_auth_from_conf_options(
conf, nova.conf.vendordata.vendordata_group.name)
if not _ADMIN_AUTH:
LOG.warning(_LW('Passing insecure dynamic vendordata requests '
'because of missing or incorrect service account '
'configuration.'))
if not _SESSION:
_SESSION = ks_loading.load_session_from_conf_options(
conf, nova.conf.vendordata.vendordata_group.name,
auth=_ADMIN_AUTH)
return _SESSION