/opt/imh-python/lib/python3.9/site-packages/libcloud/common
Edit: /opt/imh-python/lib/python3.9/site-packages/libcloud/common/openstack_identity.py (61740B)
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements. See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You 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.
"""
Common / shared code for handling authentication against OpenStack identity
service (Keystone).
"""
import datetime
from libcloud.utils.py3 import httplib
from libcloud.utils.iso8601 import parse_date
from libcloud.common.base import (ConnectionUserAndKey, Response,
CertificateConnection)
from libcloud.compute.types import (LibcloudError, InvalidCredsError,
MalformedResponseError)
try:
import simplejson as json
except ImportError:
import json # type: ignore
AUTH_API_VERSION = '1.1'
# Auth versions which contain token expiration information.
AUTH_VERSIONS_WITH_EXPIRES = [
'1.1',
'2.0',
'2.0_apikey',
'2.0_password',
'2.0_voms',
'3.0',
'3.x_password',
'3.x_oidc_access_token'
]
# How many seconds to subtract from the auth token expiration time before
# testing if the token is still valid.
# The time is subtracted to account for the HTTP request latency and prevent
# user from getting "InvalidCredsError" if token is about to expire.
AUTH_TOKEN_EXPIRES_GRACE_SECONDS = 5
__all__ = [
'OpenStackIdentityVersion',
'OpenStackIdentityDomain',
'OpenStackIdentityProject',
'OpenStackIdentityUser',
'OpenStackIdentityRole',
'OpenStackServiceCatalog',
'OpenStackServiceCatalogEntry',
'OpenStackServiceCatalogEntryEndpoint',
'OpenStackIdentityEndpointType',
'OpenStackIdentityConnection',
'OpenStackIdentity_1_0_Connection',
'OpenStackIdentity_1_1_Connection',
'OpenStackIdentity_2_0_Connection',
'OpenStackIdentity_2_0_Connection_VOMS',
'OpenStackIdentity_3_0_Connection',
'OpenStackIdentity_3_0_Connection_OIDC_access_token',
'get_class_for_auth_version'
]
class OpenStackIdentityEndpointType(object):
"""
Enum class for openstack identity endpoint type.
"""
INTERNAL = 'internal'
EXTERNAL = 'external'
ADMIN = 'admin'
class OpenStackIdentityTokenScope(object):
"""
Enum class for openstack identity token scope.
"""
PROJECT = 'project'
DOMAIN = 'domain'
UNSCOPED = 'unscoped'
class OpenStackIdentityVersion(object):
def __init__(self, version, status, updated, url):
self.version = version
self.status = status
self.updated = updated
self.url = url
def __repr__(self):
return (('
' %
(self.version, self.status, self.updated, self.url)))
class OpenStackIdentityDomain(object):
def __init__(self, id, name, enabled):
self.id = id
self.name = name
self.enabled = enabled
def __repr__(self):
return (('' %
(self.id, self.name, self.enabled)))
class OpenStackIdentityProject(object):
def __init__(self, id, name, description, enabled, domain_id=None):
self.id = id
self.name = name
self.description = description
self.enabled = enabled
self.domain_id = domain_id
def __repr__(self):
return (('' %
(self.id, self.domain_id, self.name, self.enabled)))
class OpenStackIdentityRole(object):
def __init__(self, id, name, description, enabled):
self.id = id
self.name = name
self.description = description
self.enabled = enabled
def __repr__(self):
return (('' % (self.id, self.name, self.description,
self.enabled)))
class OpenStackIdentityUser(object):
def __init__(self, id, domain_id, name, email, description, enabled):
self.id = id
self.domain_id = domain_id
self.name = name
self.email = email
self.description = description
self.enabled = enabled
def __repr__(self):
return (('' % (self.id, self.domain_id, self.name,
self.email, self.enabled)))
class OpenStackServiceCatalog(object):
"""
http://docs.openstack.org/api/openstack-identity-service/2.0/content/
This class should be instantiated with the contents of the
'serviceCatalog' in the auth response. This will do the work of figuring
out which services actually exist in the catalog as well as split them up
by type, name, and region if available
"""
_auth_version = None
_service_catalog = None
def __init__(self, service_catalog, auth_version=AUTH_API_VERSION):
self._auth_version = auth_version
# Check this way because there are a couple of different 2.0_*
# auth types.
if '3.x' in self._auth_version:
entries = self._parse_service_catalog_auth_v3(
service_catalog=service_catalog)
elif '2.0' in self._auth_version:
entries = self._parse_service_catalog_auth_v2(
service_catalog=service_catalog)
elif ('1.1' in self._auth_version) or ('1.0' in self._auth_version):
entries = self._parse_service_catalog_auth_v1(
service_catalog=service_catalog)
else:
raise LibcloudError('auth version "%s" not supported'
% (self._auth_version))
# Force consistent ordering by sorting the entries
entries = sorted(entries,
key=lambda x: x.service_type + (x.service_name or ''))
self._entries = entries # stories all the service catalog entries
def get_entries(self):
"""
Return all the entries for this service catalog.
:rtype: ``list`` of :class:`.OpenStackServiceCatalogEntry`
"""
return self._entries
def get_catalog(self):
"""
Deprecated in the favor of ``get_entries`` method.
"""
return self.get_entries()
def get_public_urls(self, service_type=None, name=None):
"""
Retrieve all the available public (external) URLs for the provided
service type and name.
"""
endpoints = self.get_endpoints(service_type=service_type,
name=name)
result = []
for endpoint in endpoints:
endpoint_type = endpoint.endpoint_type
if endpoint_type == OpenStackIdentityEndpointType.EXTERNAL:
result.append(endpoint.url)
return result
def get_endpoints(self, service_type=None, name=None):
"""
Retrieve all the endpoints for the provided service type and name.
:rtype: ``list`` of :class:`.OpenStackServiceCatalogEntryEndpoint`
"""
endpoints = []
for entry in self._entries:
# Note: "if XXX and YYY != XXX" comparison is used to support
# partial lookups.
# This allows user to pass in only one argument to the method (only
# service_type or name), both of them or neither.
if service_type and entry.service_type != service_type:
continue
if name and entry.service_name != name:
continue
for endpoint in entry.endpoints:
endpoints.append(endpoint)
return endpoints
def get_endpoint(self, service_type=None, name=None, region=None,
endpoint_type=OpenStackIdentityEndpointType.EXTERNAL):
"""
Retrieve a single endpoint using the provided criteria.
Note: If no or more than one matching endpoint is found, an exception
is thrown.
"""
endpoints = []
for entry in self._entries:
if service_type and entry.service_type != service_type:
continue
if name and entry.service_name != name:
continue
for endpoint in entry.endpoints:
if region and endpoint.region != region:
continue
if endpoint_type and endpoint.endpoint_type != endpoint_type:
continue
endpoints.append(endpoint)
if len(endpoints) == 1:
return endpoints[0]
elif len(endpoints) > 1:
raise ValueError('Found more than 1 matching endpoint')
else:
raise LibcloudError('Could not find specified endpoint')
def get_regions(self, service_type=None):
"""
Retrieve a list of all the available regions.
:param service_type: If specified, only return regions for this
service type.
:type service_type: ``str``
:rtype: ``list`` of ``str``
"""
regions = set()
for entry in self._entries:
if service_type and entry.service_type != service_type:
continue
for endpoint in entry.endpoints:
if endpoint.region:
regions.add(endpoint.region)
return sorted(list(regions))
def get_service_types(self, region=None):
"""
Retrieve all the available service types.
:param region: Optional region to retrieve service types for.
:type region: ``str``
:rtype: ``list`` of ``str``
"""
service_types = set()
for entry in self._entries:
include = True
for endpoint in entry.endpoints:
if region and endpoint.region != region:
include = False
break
if include:
service_types.add(entry.service_type)
return sorted(list(service_types))
def get_service_names(self, service_type=None, region=None):
"""
Retrieve list of service names that match service type and region.
:type service_type: ``str``
:type region: ``str``
:rtype: ``list`` of ``str``
"""
names = set()
if '2.0' not in self._auth_version:
raise ValueError('Unsupported version: %s' % (self._auth_version))
for entry in self._entries:
if service_type and entry.service_type != service_type:
continue
include = True
for endpoint in entry.endpoints:
if region and endpoint.region != region:
include = False
break
if include and entry.service_name:
names.add(entry.service_name)
return sorted(list(names))
def _parse_service_catalog_auth_v1(self, service_catalog):
entries = []
for service, endpoints in service_catalog.items():
entry_endpoints = []
for endpoint in endpoints:
region = endpoint.get('region', None)
public_url = endpoint.get('publicURL', None)
private_url = endpoint.get('internalURL', None)
if public_url:
entry_endpoint = OpenStackServiceCatalogEntryEndpoint(
region=region, url=public_url,
endpoint_type=OpenStackIdentityEndpointType.EXTERNAL)
entry_endpoints.append(entry_endpoint)
if private_url:
entry_endpoint = OpenStackServiceCatalogEntryEndpoint(
region=region, url=private_url,
endpoint_type=OpenStackIdentityEndpointType.INTERNAL)
entry_endpoints.append(entry_endpoint)
entry = OpenStackServiceCatalogEntry(service_type=service,
endpoints=entry_endpoints)
entries.append(entry)
return entries
def _parse_service_catalog_auth_v2(self, service_catalog):
entries = []
for service in service_catalog:
service_type = service['type']
service_name = service.get('name', None)
entry_endpoints = []
for endpoint in service.get('endpoints', []):
region = endpoint.get('region', None)
public_url = endpoint.get('publicURL', None)
private_url = endpoint.get('internalURL', None)
if public_url:
entry_endpoint = OpenStackServiceCatalogEntryEndpoint(
region=region, url=public_url,
endpoint_type=OpenStackIdentityEndpointType.EXTERNAL)
entry_endpoints.append(entry_endpoint)
if private_url:
entry_endpoint = OpenStackServiceCatalogEntryEndpoint(
region=region, url=private_url,
endpoint_type=OpenStackIdentityEndpointType.INTERNAL)
entry_endpoints.append(entry_endpoint)
entry = OpenStackServiceCatalogEntry(service_type=service_type,
endpoints=entry_endpoints,
service_name=service_name)
entries.append(entry)
return entries
def _parse_service_catalog_auth_v3(self, service_catalog):
entries = []
for item in service_catalog:
service_type = item['type']
service_name = item.get('name', None)
entry_endpoints = []
for endpoint in item['endpoints']:
region = endpoint.get('region', None)
url = endpoint['url']
endpoint_type = endpoint['interface']
if endpoint_type == 'internal':
endpoint_type = OpenStackIdentityEndpointType.INTERNAL
elif endpoint_type == 'public':
endpoint_type = OpenStackIdentityEndpointType.EXTERNAL
elif endpoint_type == 'admin':
endpoint_type = OpenStackIdentityEndpointType.ADMIN
entry_endpoint = OpenStackServiceCatalogEntryEndpoint(
region=region, url=url, endpoint_type=endpoint_type)
entry_endpoints.append(entry_endpoint)
entry = OpenStackServiceCatalogEntry(service_type=service_type,
service_name=service_name,
endpoints=entry_endpoints)
entries.append(entry)
return entries
class OpenStackServiceCatalogEntry(object):
def __init__(self, service_type, endpoints=None, service_name=None):
"""
:param service_type: Service type.
:type service_type: ``str``
:param endpoints: Endpoints belonging to this entry.
:type endpoints: ``list``
:param service_name: Optional service name.
:type service_name: ``str``
"""
self.service_type = service_type
self.endpoints = endpoints or []
self.service_name = service_name
# For consistency, sort the endpoints
self.endpoints = sorted(self.endpoints, key=lambda x: x.url or '')
def __eq__(self, other):
return (self.service_type == other.service_type and
self.endpoints == other.endpoints and
other.service_name == self.service_name)
def __ne__(self, other):
return not self.__eq__(other=other)
def __repr__(self):
return (('