[WIP]add new signer for dlf

docs/content/concepts/rest/dlf.md

@@ -115,3 +115,59 @@ WITH (
     -- 'dlf.token-ecs-role-name' = 'my_ecs_role_name'
 );
 ```
+
+## Signing Algorithm Configuration
+
+Paimon supports multiple signing algorithms for DLF authentication. You can configure the signing algorithm explicitly,
+or let Paimon automatically select it based on the endpoint host.
+
+### Automatic Selection (Recommended)
+
+By default, Paimon automatically selects the appropriate signing algorithm based on the endpoint URI:
+
+- **DLF endpoints** (e.g., `cn-hangzhou-vpc.dlf.aliyuncs.com`): Automatically uses `dlf-default`
+  (backward compatible). Recommended for VPC environments with better performance.
+- **OpenAPI endpoints** (e.g., `dlfnext.cn-hangzhou.aliyuncs.com`): Automatically uses
+  `dlf-openapi` for DlfNext/2025-03-10 OpenAPI. Supports public network access through Alibaba Cloud API infrastructure
+   for special scenarios.
+
+```sql
+CREATE CATALOG `paimon-rest-catalog`
+WITH (
+    'type' = 'paimon',
+    'uri' = 'https://dlfnext.cn-hangzhou.aliyuncs.com',  -- Auto-detected as dlf-openapi
+    'metastore' = 'rest',
+    'warehouse' = 'my_instance_name',
+    'token.provider' = 'dlf',
+    'dlf.access-key-id'='<access-key-id>',
+    'dlf.access-key-secret'='<access-key-secret>'
+    -- 'dlf.signing-algorithm' is not set, will be auto-detected
+);
+```
+
+### Explicit Configuration
+
+You can explicitly specify the signing algorithm:
+
+```sql
+CREATE CATALOG `paimon-rest-catalog`
+WITH (
+    'type' = 'paimon',
+    'uri' = '<catalog server url>',
+    'metastore' = 'rest',
+    'warehouse' = 'my_instance_name',
+    'token.provider' = 'dlf',
+    'dlf.access-key-id'='<access-key-id>',
+    'dlf.access-key-secret'='<access-key-secret>',
+    'dlf.signing-algorithm' = 'dlf-default'  -- or 'dlf-openapi'
+);
+```
+
+**Available signing algorithms:**
+
+- `dlf-default` (default): DLF4-HMAC-SHA256 signer for default VPC endpoint, backward compatible
+  with existing DLF authentication
+- `dlf-openapi`: ROA v2 style signer for DlfNext/2025-03-10 OpenAPI, implements HMAC-SHA1
+  signature with ROA style canonicalization
+
+**Note:** When `dlf.signing-algorithm` is explicitly configured, it takes precedence over automatic detection.

paimon-api/src/main/java/org/apache/paimon/rest/RESTCatalogOptions.java

@@ -104,6 +104,15 @@ public class RESTCatalogOptions {
                     .noDefaultValue()
                     .withDescription("REST Catalog DLF OSS endpoint.");
 
+    public static final ConfigOption<String> DLF_SIGNING_ALGORITHM =
+            ConfigOptions.key("dlf.signing-algorithm")
+                    .stringType()
+                    .defaultValue("dlf-default")
+                    .withDescription(
+                            "DLF signing algorithm. Options: 'dlf-default' (for default VPC endpoint), "
+                                    + "'dlf-openapi' (for DlfNext/2025-03-10). "
+                                    + "If not set, will be automatically selected based on endpoint host.");
+
     public static final ConfigOption<Boolean> IO_CACHE_ENABLED =
             ConfigOptions.key("io-cache.enabled")
                     .booleanType()

paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFAuthProvider.java

@@ -25,8 +25,7 @@
 
 import javax.annotation.Nullable;
 
-import java.time.ZoneOffset;
-import java.time.ZonedDateTime;
+import java.time.Instant;
 import java.time.format.DateTimeFormatter;
 import java.util.HashMap;
 import java.util.Map;
@@ -53,49 +52,76 @@ public class DLFAuthProvider implements AuthProvider {
     protected static final String MEDIA_TYPE = "application/json";
 
     @Nullable private final DLFTokenLoader tokenLoader;
+    private final String uri;
     private final String region;
+    private final String signingAlgorithm;
 
     @Nullable protected volatile DLFToken token;
+    private final DLFRequestSigner signer;
 
-    public static DLFAuthProvider fromTokenLoader(DLFTokenLoader tokenLoader, String region) {
-        return new DLFAuthProvider(tokenLoader, null, region);
+    public static DLFAuthProvider fromTokenLoader(
+            DLFTokenLoader tokenLoader, String uri, String region, String signingAlgorithm) {
+        return new DLFAuthProvider(tokenLoader, null, uri, region, signingAlgorithm);
     }
 
     public static DLFAuthProvider fromAccessKey(
-            String accessKeyId, String accessKeySecret, String securityToken, String region) {
+            String accessKeyId,
+            String accessKeySecret,
+            @Nullable String securityToken,
+            String uri,
+            String region,
+            String signingAlgorithm) {
         DLFToken token = new DLFToken(accessKeyId, accessKeySecret, securityToken, null);
-        return new DLFAuthProvider(null, token, region);
+        return new DLFAuthProvider(null, token, uri, region, signingAlgorithm);
     }
 
     public DLFAuthProvider(
-            @Nullable DLFTokenLoader tokenLoader, @Nullable DLFToken token, String region) {
+            @Nullable DLFTokenLoader tokenLoader,
+            @Nullable DLFToken token,
+            String uri,
+            String region,
+            String signingAlgorithm) {
         this.tokenLoader = tokenLoader;
         this.token = token;
+        this.uri = uri;
         this.region = region;
+        this.signingAlgorithm = signingAlgorithm;
+        this.signer = createSigner(signingAlgorithm);
     }
 
     @Override
     public Map<String, String> mergeAuthHeader(
             Map<String, String> baseHeader, RESTAuthParameter restAuthParameter) {
         DLFToken token = getFreshToken();
         try {
-            String dateTime =
-                    baseHeader.getOrDefault(
-                            DLF_DATE_HEADER_KEY.toLowerCase(),
-                            ZonedDateTime.now(ZoneOffset.UTC).format(AUTH_DATE_TIME_FORMATTER));
-            String date = dateTime.substring(0, 8);
+            Instant now = Instant.now();
             Map<String, String> signHeaders =
-                    generateSignHeaders(
-                            restAuthParameter.data(), dateTime, token.getSecurityToken());
-            String authorization =
-                    DLFAuthSignature.getAuthorization(
-                            restAuthParameter, token, region, signHeaders, dateTime, date);
+                    signer.signHeaders(
+                            restAuthParameter.data(), now, token.getSecurityToken(), uri);
+            String authorization = signer.authorization(restAuthParameter, token, uri, signHeaders);
             Map<String, String> headersWithAuth = new HashMap<>(baseHeader);
             headersWithAuth.putAll(signHeaders);
             headersWithAuth.put(DLF_AUTHORIZATION_HEADER_KEY, authorization);
             return headersWithAuth;
         } catch (Exception e) {
-            throw new RuntimeException(e);
+            throw new RuntimeException("Failed to generate authorization header", e);
+        }
+    }
+
+    private DLFRequestSigner createSigner(String signingAlgorithm) {
+        switch (signingAlgorithm) {
+            case DLFDefaultSigner.IDENTIFIER:
+                return new DLFDefaultSigner(region);
+            case DLFOpenApiSigner.IDENTIFIER:
+                return new DLFOpenApiSigner();
+            default:
+                throw new IllegalArgumentException(
+                        "Unknown DLF signing algorithm: "
+                                + signingAlgorithm
+                                + ". Supported: "
+                                + DLFDefaultSigner.IDENTIFIER
+                                + ", "
+                                + DLFOpenApiSigner.IDENTIFIER);
         }
     }
 
@@ -135,20 +161,4 @@ private boolean shouldRefresh() {
         long now = System.currentTimeMillis();
         return expireTime - now < TOKEN_EXPIRATION_SAFE_TIME_MILLIS;
     }
-
-    public static Map<String, String> generateSignHeaders(
-            String data, String dateTime, String securityToken) throws Exception {
-        Map<String, String> signHeaders = new HashMap<>();
-        signHeaders.put(DLF_DATE_HEADER_KEY, dateTime);
-        signHeaders.put(DLF_CONTENT_SHA56_HEADER_KEY, DLF_CONTENT_SHA56_VALUE);
-        signHeaders.put(DLF_AUTH_VERSION_HEADER_KEY, DLFAuthSignature.VERSION);
-        if (data != null && !data.isEmpty()) {
-            signHeaders.put(DLF_CONTENT_TYPE_KEY, MEDIA_TYPE);
-            signHeaders.put(DLF_CONTENT_MD5_HEADER_KEY, DLFAuthSignature.md5(data));
-        }
-        if (securityToken != null) {
-            signHeaders.put(DLF_SECURITY_TOKEN_HEADER_KEY, securityToken);
-        }
-        return signHeaders;
-    }
 }

paimon-api/src/main/java/org/apache/paimon/rest/auth/DLFAuthProviderFactory.java

@@ -20,6 +20,7 @@
 
 import org.apache.paimon.options.Options;
 import org.apache.paimon.rest.RESTCatalogOptions;
+import org.apache.paimon.utils.StringUtils;
 
 import java.util.regex.Matcher;
 import java.util.regex.Pattern;
@@ -36,25 +37,32 @@ public String identifier() {
 
     @Override
     public AuthProvider create(Options options) {
+        String uri = options.get(URI);
         String region =
                 options.getOptional(RESTCatalogOptions.DLF_REGION)
-                        .orElseGet(() -> parseRegionFromUri(options.get(URI)));
+                        .orElseGet(() -> parseRegionFromUri(uri));
+        String signingAlgorithm =
+                options.getOptional(RESTCatalogOptions.DLF_SIGNING_ALGORITHM)
+                        .orElseGet(() -> parseSigningAlgoFromUri(uri));
+
         if (options.getOptional(RESTCatalogOptions.DLF_TOKEN_LOADER).isPresent()) {
             DLFTokenLoader dlfTokenLoader =
                     DLFTokenLoaderFactory.createDLFTokenLoader(
                             options.get(RESTCatalogOptions.DLF_TOKEN_LOADER), options);
-            return DLFAuthProvider.fromTokenLoader(dlfTokenLoader, region);
+            return DLFAuthProvider.fromTokenLoader(dlfTokenLoader, uri, region, signingAlgorithm);
         } else if (options.getOptional(RESTCatalogOptions.DLF_TOKEN_PATH).isPresent()) {
             DLFTokenLoader dlfTokenLoader =
                     DLFTokenLoaderFactory.createDLFTokenLoader("local_file", options);
-            return DLFAuthProvider.fromTokenLoader(dlfTokenLoader, region);
+            return DLFAuthProvider.fromTokenLoader(dlfTokenLoader, uri, region, signingAlgorithm);
         } else if (options.getOptional(RESTCatalogOptions.DLF_ACCESS_KEY_ID).isPresent()
                 && options.getOptional(RESTCatalogOptions.DLF_ACCESS_KEY_SECRET).isPresent()) {
             return DLFAuthProvider.fromAccessKey(
                     options.get(RESTCatalogOptions.DLF_ACCESS_KEY_ID),
                     options.get(RESTCatalogOptions.DLF_ACCESS_KEY_SECRET),
                     options.get(RESTCatalogOptions.DLF_SECURITY_TOKEN),
-                    region);
+                    uri,
+                    region,
+                    signingAlgorithm);
         }
         throw new IllegalArgumentException("DLF token path or AK must be set for DLF Auth.");
     }
@@ -74,4 +82,25 @@ protected static String parseRegionFromUri(String uri) {
         throw new IllegalArgumentException(
                 "Could not get region from conf or uri, please check your config.");
     }
+
+    /**
+     * Parse signing algorithm from uri. Automatically selects the appropriate signer based on the
+     * endpoint uri.
+     *
+     * @param uri endpoint uri
+     * @return signing algorithm identifier
+     */
+    protected static String parseSigningAlgoFromUri(String uri) {
+        if (StringUtils.isEmpty(uri)) {
+            return DLFDefaultSigner.IDENTIFIER;
+        }
+
+        // Check for aliyun openapi endpoints
+        if (uri.toLowerCase().contains("dlfnext")) {
+            return DLFOpenApiSigner.IDENTIFIER;
+        }
+
+        // Default to dlf for unknown hosts
+        return DLFDefaultSigner.IDENTIFIER;
+    }
 }