Merge 'feature/support_rds_auth' into 'master'

chore: add rds connect utils

See merge request: !1055

Author: volc-sdk-team

volcenginesdkcore/feature/__init__.py

@@ -0,0 +1,6 @@
+# coding: utf-8
+"""
+Feature modules for specific service utilities.
+
+This package contains specialized utility modules for various Volcengine services.
+"""

volcenginesdkcore/feature/rds/__init__.py

@@ -0,0 +1,4 @@
+# coding: utf-8
+from volcenginesdkcore.feature.rds.connect_utils import build_auth_token
+
+__all__ = ['build_auth_token']

volcenginesdkcore/feature/rds/connect_utils.py

@@ -0,0 +1,80 @@
+# coding: utf-8
+"""
+RDS MySQL connection authentication utilities.
+
+This module provides utilities for generating authentication tokens for RDS MySQL database connections.
+"""
+
+from volcenginesdkcore.endpoint.providers.standard_provider import StandardEndpointResolver
+from volcenginesdkcore.interceptor import InterceptorChain, InterceptorContext, SignRequestInterceptor, \
+    ResolveEndpointInterceptor
+from volcenginesdkcore.interceptor import Request
+
+DEFAULT_SERVICE = 'rds_mysql'
+DEFAULT_API_VERSION = '2022-01-01'
+DEFAULT_API = 'ConnectDatabase'
+DEFAULT_EXPIRES = 900
+
+
+def build_auth_token(api_client, db_user, instance_id, expires=None):
+    """
+    Build an authentication token (presigned URL) for connecting to RDS MySQL database.
+
+    :param api_client: ApiClient instance
+    :param db_user: Database username
+    :param instance_id: RDS instance ID
+    :param expires: Token expiration time in seconds (default: 900, i.e., 15 minutes)
+    :return: Presigned URL string for database authentication
+    :raises ValueError: If required parameters are missing or invalid
+    """
+    # Validate api_client
+    if api_client is None:
+        raise ValueError("api_client must not be None")
+
+    configuration = api_client.configuration
+    region = configuration.region
+
+    # Validate inputs
+    if not region:
+        raise ValueError("region must not be empty")
+
+    if not db_user:
+        raise ValueError("db_user must not be empty")
+
+    if not instance_id:
+        raise ValueError("instance_id must not be empty")
+
+    # Set default expiration time
+    if expires is None or expires <= 0:
+        expires = DEFAULT_EXPIRES
+
+    # Build query parameters
+    query = {
+        'Action': DEFAULT_API,
+        'Version': DEFAULT_API_VERSION,
+        'X-Expires': str(expires),
+        'DBUser': db_user,
+        'InstanceId': instance_id,
+    }
+
+    # Create Request with presign mode
+    request = Request(configuration,
+                      resource_path='/{}/{}/{}/get/text_plain/'.format(DEFAULT_API, DEFAULT_API_VERSION,
+                                                                       DEFAULT_SERVICE),
+                      method='GET',
+                      query_params=query)
+    request.endpoint_provider = StandardEndpointResolver()
+    request.service = DEFAULT_SERVICE
+    request.is_presign = True
+
+    # Create interceptor chain:
+    #   ResolveEndpointInterceptor - resolves endpoint + scheme
+    #   SignRequestInterceptor     - presign URL signing
+    chain = InterceptorChain()
+    chain.append_request_interceptor(ResolveEndpointInterceptor())
+    chain.append_request_interceptor(SignRequestInterceptor())
+
+    context = InterceptorContext(request=request)
+    context = chain.execute_request(context)
+
+    return '{url}?{query}'.format(url=context.request.url, query=context.request.signed_query)

volcenginesdkcore/interceptor/interceptors/request.py

@@ -63,6 +63,10 @@ def __init__(
         self.retryer = configuration.retryer
         self.credential_provider = configuration.credential_provider
 
+        # Presign support
+        self.is_presign = False
+        self.signed_query = None
+
         self.runtime_options = None
         if hasattr(body, '_configuration') and isinstance(body._configuration, RuntimeOption):
             self.runtime_options = body._configuration

volcenginesdkcore/interceptor/interceptors/resolve_endpoint_interceptor.py

@@ -19,7 +19,14 @@ def intercept(self, context):
             context.request.host = endpoint_resolver.host
             prefix = endpoint_resolver.url_for(scheme)
         else:
-            prefix = scheme + '://' + host
+            if host.startswith('https://'):
+                prefix = host
+                context.request.host = host[len('https://'):]
+            elif host.startswith('http://'):
+                prefix = host
+                context.request.host = host[len('http://'):]
+            else:
+                prefix = scheme + '://' + host
         context.request.url = prefix + context.request.true_path
         sdk_core_logger.debug_endpoint(
             "Using endpoint: %s", context.request.host

volcenginesdkcore/interceptor/interceptors/sign_request_interceptor.py

@@ -13,23 +13,36 @@ def name(self):
     def intercept(self, context):
         # 新增代码。处理assume_role和assume_role_oidc和assume_role_saml
         if context.request.credential_provider is not None:
-            credentials = context.request.credential_provider.get_credentials() # 这会调用 _assume_role_oidc() 方法获取临时凭证
+            credentials = context.request.credential_provider.get_credentials()  # 这会调用 _assume_role_oidc() 方法获取临时凭证
             context.request.ak = credentials.ak
             context.request.sk = credentials.sk
             context.request.session_token = credentials.session_token
 
-        self.update_params_for_auth(host=context.request.host, path=context.request.true_path,
-                                    method=context.request.method,
-                                    headers=context.request.header_params,
-                                    querys=context.request.query_params,
-                                    auth_settings=context.request.auth_settings,
-                                    body=context.request.body,
-                                    post_params=context.request.post_params,
-                                    service=context.request.service,
-                                    ak=context.request.ak,
-                                    sk=context.request.sk,
-                                    session_token=context.request.session_token,
-                                    region=context.request.region)
+        if context.request.is_presign:
+            context.request.signed_query = SignerV4.sign_url(
+                path=context.request.true_path,
+                method=context.request.method,
+                query=context.request.query_params,
+                ak=context.request.ak,
+                sk=context.request.sk,
+                region=context.request.region,
+                service=context.request.service,
+                session_token=context.request.session_token,
+                host=context.request.host,
+            )
+        else:
+            self.update_params_for_auth(host=context.request.host, path=context.request.true_path,
+                                        method=context.request.method,
+                                        headers=context.request.header_params,
+                                        querys=context.request.query_params,
+                                        auth_settings=context.request.auth_settings,
+                                        body=context.request.body,
+                                        post_params=context.request.post_params,
+                                        service=context.request.service,
+                                        ak=context.request.ak,
+                                        sk=context.request.sk,
+                                        session_token=context.request.session_token,
+                                        region=context.request.region)
         return context
 
     @staticmethod

volcenginesdkcore/signv4.py

@@ -91,3 +91,89 @@ def get_signing_secret_key_v4(sk, date, region, service):
     @staticmethod
     def hmac_sha256(key, msg):
         return hmac.new(key, msg.encode('utf-8'), hashlib.sha256).digest()
+
+    @staticmethod
+    def sign_url(path, method, query, ak, sk, region, service, session_token=None, host=None):
+        """
+        Generate presigned URL query string (AWS Signature V4)
+
+        :param path: Request path
+        :param method: HTTP method (GET, POST, etc.)
+        :param query: Query parameters dict
+        :param ak: Access Key
+        :param sk: Secret Key
+        :param region: Service region
+        :param service: Service name
+        :param session_token: Optional session token
+        :param host: Optional host header to sign
+        :return: Query string with signature
+        """
+        format_date = datetime.datetime.utcnow().strftime("%Y%m%dT%H%M%SZ")
+        date = format_date[:8]
+
+        # Build credential scope
+        credential_scope = '/'.join([date, region, service, 'request'])
+
+        # Determine if host header should be signed
+        sign_host = host is not None and host != ''
+
+        # Add required query parameters
+        query = dict(query)  # Make a copy to avoid modifying original
+        query['X-Date'] = format_date
+        query['X-NotSignBody'] = ''
+        query['X-Credential'] = ak + '/' + credential_scope
+        query['X-Algorithm'] = 'HMAC-SHA256'
+        query['X-SignedHeaders'] = 'host' if sign_host else ''
+        query['X-SignedQueries'] = ''
+
+        # Generate X-SignedQueries BEFORE adding X-Security-Token
+        query['X-SignedQueries'] = ';'.join(sorted(query.keys()))
+        signed_query_keys = set(query.keys())
+
+        # X-Security-Token must be added AFTER X-SignedQueries calculation
+        if session_token:
+            query['X-Security-Token'] = session_token
+
+        # Build canonical request
+        body_hash = hashlib.sha256(b'').hexdigest()
+        canonical_query_params = {k: v for k, v in query.items() if k in signed_query_keys}
+
+        if sign_host:
+            canonical_request = '\n'.join([
+                method,
+                path,
+                SignerV4.canonical_query(canonical_query_params),
+                'host:' + host + '\n',
+                'host',
+                body_hash
+            ])
+        else:
+            canonical_request = '\n'.join([
+                method,
+                path,
+                SignerV4.canonical_query(canonical_query_params),
+                '\n',
+                '',
+                body_hash
+            ])
+        sdk_core_logger.debug_sign("[sign_url] canonical_request:\n%s", canonical_request)
+
+        # Build string to sign
+        signing_str = '\n'.join([
+            'HMAC-SHA256',
+            format_date,
+            credential_scope,
+            hashlib.sha256(canonical_request.encode('utf-8')).hexdigest()
+        ])
+        sdk_core_logger.debug_sign("[sign_url] string_to_sign:\n%s", signing_str)
+
+        # Calculate signature
+        signing_key = SignerV4.get_signing_secret_key_v4(sk, date, region, service)
+        signature = hmac.new(signing_key, signing_str.encode('utf-8'), hashlib.sha256).hexdigest()
+        sdk_core_logger.debug_sign("[sign_url] calculated signature: %s", signature)
+
+        # Add signature to query
+        query['X-Signature'] = signature
+
+        # Return encoded query string
+        return urlencode(sorted(query.items()))