AKSK认证时候支持客户自定义auth url

Author: karonganyong

PKG-INFO

@@ -1,6 +1,6 @@
 Metadata-Version: 1.1
 Name: huaweicloud-sdk-python
-Version: 1.0.20
+Version: 1.0.21
 Summary: An SDK for building applications to work with OpenStack
 Home-page: https://github.com/huaweicloud/huaweicloud-sdk-python
 Author: huawei

openstack/aksksession.py

@@ -26,7 +26,8 @@
 from openstack.exceptions import EndpointNotFound, SDKException
 from openstack.session import DEFAULT_USER_AGENT
 from openstack import session as osession
-from keystoneauth1 import _utils as utils
+from keystoneauth1 import _utils as log_utils
+from openstack import utils
 from openstack.session import map_exceptions
 from openstack.service_endpoint import endpoint as _endpoint
 from keystoneauth1 import exceptions
@@ -38,9 +39,10 @@
 ALGORITHM = "SDK-HMAC-SHA256"
 PYTHON2 = "2"
 UTF8 = "utf-8"
-IAMURL = "https://iam.%s/v3/%s"
+PROJECTIAMURL = "https://iam.%s.%s/v3/"
+GLOBALIAMURL = "https://iam.%s/v3/"
 
-_logger = utils.get_logger(__name__)
+_logger = log_utils.get_logger(__name__)
 
 
 def construct_session(session_obj=None):
@@ -152,7 +154,7 @@ def _make_canonical_request(self, method=None, url=None, headers=None, params=No
         canonical_header = '\n'.join(canonical_header)
         canonical_header += '\n'
         signed_header = ';'.join([k.lower() for k in self.headtosign])
-        body = body if body  else ""
+        body = body if body else ""
         request_payload = hashlib.sha256(get_utf8_bytes(body)).hexdigest()
         return '\n'.join(
             [canonical_method, canonical_uri, canonical_querystring, canonical_header, signed_header, request_payload])
@@ -254,6 +256,7 @@ def __init__(self, profile,
                  **kwargs
                  ):
         self.project_id = kwargs.get("project_id")
+        self.auth_url = kwargs.get('auth_url', None)
         self.domain_id = kwargs.get("domain_id", None)
         self.domain = kwargs.get("domain")
         self.region = kwargs.get("region")
@@ -491,10 +494,14 @@ def _get_endpoint_from_configdata(self, service_type, interface):
 
     def __fetch_all_endpoint_service_project_level(self):
         kvendpoints = {}
-        resp = self.request(IAMURL % (self.domain, "endpoints"), "GET",
+        if self.auth_url:
+            auth_url = self.auth_url
+        else:
+            auth_url = PROJECTIAMURL % (self.region, self.domain)
+        resp = self.request(utils.urljoin(auth_url, "endpoints"), "GET",
                             endpoint_filter=identity_service.AdminService())
         endpoints = resp.json().get("endpoints", [])
-        resp = self.request(IAMURL % (self.domain, "services"), "GET",
+        resp = self.request(utils.urljoin(auth_url, "services"), "GET",
                             endpoint_filter=identity_service.AdminService())
         services = resp.json().get("services", [])
         id_endpoint_map, servicetype_id_map = {}, {}
@@ -520,10 +527,14 @@ def __fetch_all_endpoint_service_project_level(self):
 
     def __fetch_all_endpoint_service_global_level(self):
         kvendpoints = {}
-        resp = self.request(IAMURL % (self.domain, "endpoints"), "GET",
+        if self.auth_url:
+            auth_url = self.auth_url
+        else:
+            auth_url = GLOBALIAMURL % self.domain
+        resp = self.request(utils.urljoin(auth_url, "endpoints"), "GET",
                             endpoint_filter=identity_service.AdminService())
         endpoints = resp.json().get("endpoints", [])
-        resp = self.request(IAMURL % (self.domain, "services"), "GET",
+        resp = self.request(utils.urljoin(auth_url, "services"), "GET",
                             endpoint_filter=identity_service.AdminService())
         services = resp.json().get("services", [])
         id_endpoint_map, servicetype_id_map = {}, {}

openstack/connection.py

@@ -267,6 +267,7 @@ def __init__(self, session=None, authenticator=None, profile=None,
                                                 region=auth_args.get("region", None),
                                                 domain=auth_args.get('domain', None),
                                                 securitytoken=auth_args.get("securitytoken", None),
+                                                auth_url=auth_args.get('auth_url', None),
                                                 domain_id=auth_args.get("domain_id", None)
                                                 )
         elif auth_args.get('auth_token', None):