chore: add rds connect utils

Author: sdk-liuzhaoliang

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,76 @@
+# coding: utf-8
+"""
+RDS MySQL connection authentication utilities.
+
+This module provides utilities for generating authentication tokens for RDS MySQL database connections.
+"""
+
+from volcenginesdkcore.signv4 import SignerV4
+from volcenginesdkcore.endpoint.providers.default_provider import DefaultEndpointProvider
+
+
+def build_auth_token(credentials, db_user, instance_id, region, expires=None):
+    """
+    Build an authentication token (presigned URL) for connecting to RDS MySQL database.
+
+    :param credentials: CredentialValue object with ak, sk, and optional session_token
+    :param db_user: Database username
+    :param instance_id: RDS instance ID
+    :param region: Service region (e.g., 'cn-beijing')
+    :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 inputs
+    if credentials is None:
+        raise ValueError("credentials must not be None")
+
+    if not hasattr(credentials, 'ak') or not credentials.ak:
+        raise ValueError("credentials.ak must not be empty")
+
+    if not hasattr(credentials, 'sk') or not credentials.sk:
+        raise ValueError("credentials.sk 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")
+
+    if not region:
+        raise ValueError("region must not be empty")
+
+    # Set default expiration time
+    if expires is None or expires <= 0:
+        expires = 900  # 15 minutes
+
+    # Service configuration
+    service = 'rds_mysql'
+
+    # Get endpoint
+    endpoint_provider = DefaultEndpointProvider()
+    resolved_endpoint = endpoint_provider.endpoint_for(service, region)
+    host = resolved_endpoint.host
+
+    # Build query parameters
+    query = {
+        'Action': 'ConnectDatabase',
+        'Version': '2022-01-01',
+        'X-Expires': str(expires),
+        'DBUser': db_user,
+        'InstanceId': instance_id,
+    }
+
+    # Sign the URL
+    signed_query = SignerV4.sign_url(
+        path='/',
+        method='GET',
+        query=query,
+        ak=credentials.ak,
+        sk=credentials.sk,
+        region=region,
+        service=service,
+        session_token=getattr(credentials, 'session_token', None)
+    )
+
+    return signed_query

volcenginesdkcore/signv4.py

@@ -91,3 +91,71 @@ 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):
+        """
+        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
+        :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'])
+
+        # 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'] = ''
+        query['X-SignedQueries'] = ''
+
+        if session_token:
+            query['X-Security-Token'] = session_token
+
+        # Generate X-SignedQueries with all query parameter names
+        query['X-SignedQueries'] = ';'.join(sorted(query.keys()))
+
+        # Build canonical request with empty body
+        body_hash = hashlib.sha256(b'').hexdigest()
+        canonical_request = '\n'.join([
+            method,
+            path,
+            SignerV4.canonical_query(query),
+            '\n',  # Empty line for signed headers section
+            '',    # Empty signed headers list
+            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()))