This is an automated email from the ASF dual-hosted git repository.
JingsongLi pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/paimon.git
The following commit(s) were added to refs/heads/master by this push:
new 507a1211a6 [python] Support DLF token path authentication (#9089)
507a1211a6 is described below
commit 507a1211a62d1fd9c97e200a6f4326e5a9a9f0b0
Author: Joey <[email protected]>
AuthorDate: Fri Aug 7 22:52:04 2026 +0800
[python] Support DLF token path authentication (#9089)
---
paimon-python/pypaimon/api/auth/factory.py | 10 +-
paimon-python/pypaimon/api/token_loader.py | 55 ++++++-
paimon-python/pypaimon/common/options/config.py | 2 +
.../pypaimon/tests/rest/token_loader_test.py | 172 +++++++++++++++++++++
4 files changed, 236 insertions(+), 3 deletions(-)
diff --git a/paimon-python/pypaimon/api/auth/factory.py
b/paimon-python/pypaimon/api/auth/factory.py
index e9c92e420b..815ca12635 100644
--- a/paimon-python/pypaimon/api/auth/factory.py
+++ b/paimon-python/pypaimon/api/auth/factory.py
@@ -94,11 +94,17 @@ class AuthProviderFactory:
# Auto-detect based on URI
signing_algorithm =
DLFAuthProviderFactory.parse_signing_algo_from_uri(uri)
+ token_loader = DLFTokenLoaderFactory.create_token_loader(options)
+ token = (
+ DLFToken.from_options(options)
+ if token_loader is None
+ else None
+ )
return DLFAuthProvider(
uri=uri,
region=region,
signing_algorithm=signing_algorithm,
- token=DLFToken.from_options(options),
- token_loader=DLFTokenLoaderFactory.create_token_loader(options)
+ token=token,
+ token_loader=token_loader
)
raise ValueError('Unknown auth provider')
diff --git a/paimon-python/pypaimon/api/token_loader.py
b/paimon-python/pypaimon/api/token_loader.py
index 65aad3677a..f76eca4306 100644
--- a/paimon-python/pypaimon/api/token_loader.py
+++ b/paimon-python/pypaimon/api/token_loader.py
@@ -15,6 +15,7 @@
# specific language governing permissions and limitations
# under the License.
+import time
from abc import ABC, abstractmethod
from dataclasses import dataclass
from datetime import datetime, timezone
@@ -84,6 +85,49 @@ class DLFTokenLoader(ABC):
pass
+class DLFLocalFileTokenLoader(DLFTokenLoader):
+ """Load temporary DLF credentials from a local JSON file."""
+
+ DEFAULT_MAX_RETRIES = 5
+
+ def __init__(self, token_file_path: str):
+ self.token_file_path = token_file_path
+
+ def load_token(self) -> DLFToken:
+ return self.read_token(self.token_file_path)
+
+ def description(self) -> str:
+ return self.token_file_path
+
+ @staticmethod
+ def read_token(token_file_path: str,
+ max_retries: int = DEFAULT_MAX_RETRIES) -> DLFToken:
+ retry = 1
+ last_exception = None
+ while retry <= max_retries:
+ try:
+ with open(token_file_path, "r", encoding="utf-8") as
token_file:
+ token_json = token_file.read()
+ except Exception as e:
+ last_exception = RuntimeError(
+ "Failed to read token file: {}".format(token_file_path)
+ )
+ last_exception.__cause__ = e
+ else:
+ try:
+ return JSON.from_json(token_json, DLFToken)
+ except Exception:
+ # The file contains AK/SK/STS. Do not retain the JSON
+ # parser exception because it can include source content.
+ last_exception = RuntimeError("Failed to parse token
file.")
+
+ if retry < max_retries:
+ time.sleep(retry)
+ retry += 1
+
+ raise last_exception
+
+
class HTTPClient:
"""HTTP client with retry and timeout configuration"""
@@ -207,7 +251,7 @@ class DLFTokenLoaderFactory:
@staticmethod
def create_token_loader(options: Options) -> Optional['DLFTokenLoader']:
- """Create ECS token loader"""
+ """Create the configured token loader."""
loader = options.get(CatalogOptions.DLF_TOKEN_LOADER)
if loader == 'ecs':
ecs_metadata_url = options.get(
@@ -216,4 +260,13 @@ class DLFTokenLoaderFactory:
)
role_name = options.get(CatalogOptions.DLF_TOKEN_ECS_ROLE_NAME)
return DLFECSTokenLoader(ecs_metadata_url, role_name)
+ if loader == 'local_file':
+ return DLFLocalFileTokenLoader(
+ options.get(CatalogOptions.DLF_TOKEN_PATH)
+ )
+ if loader is not None:
+ raise ValueError("Unknown DLF token loader: {}".format(loader))
+ token_path = options.get(CatalogOptions.DLF_TOKEN_PATH)
+ if token_path is not None:
+ return DLFLocalFileTokenLoader(token_path)
return None
diff --git a/paimon-python/pypaimon/common/options/config.py
b/paimon-python/pypaimon/common/options/config.py
index 2573782c42..5684ab62c3 100644
--- a/paimon-python/pypaimon/common/options/config.py
+++ b/paimon-python/pypaimon/common/options/config.py
@@ -101,6 +101,8 @@ class CatalogOptions:
"dlf.access-key-secret").string_type().no_default_value().with_description("DLF
access key secret")
DLF_ACCESS_SECURITY_TOKEN = ConfigOptions.key(
"dlf.security-token").string_type().no_default_value().with_description("DLF
security token")
+ DLF_TOKEN_PATH =
ConfigOptions.key("dlf.token-path").string_type().no_default_value().with_description(
+ "DLF token file path")
DLF_OSS_ENDPOINT =
ConfigOptions.key("dlf.oss-endpoint").string_type().no_default_value().with_description(
"DLF OSS endpoint")
DLF_TOKEN_LOADER =
ConfigOptions.key("dlf.token-loader").string_type().no_default_value().with_description(
diff --git a/paimon-python/pypaimon/tests/rest/token_loader_test.py
b/paimon-python/pypaimon/tests/rest/token_loader_test.py
new file mode 100644
index 0000000000..1fb1d29f61
--- /dev/null
+++ b/paimon-python/pypaimon/tests/rest/token_loader_test.py
@@ -0,0 +1,172 @@
+# 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.
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from pypaimon.api.auth.factory import AuthProviderFactory
+from pypaimon.api.token_loader import (
+ DLFLocalFileTokenLoader,
+ DLFToken,
+ DLFTokenLoaderFactory,
+)
+from pypaimon.common.json_util import JSON
+from pypaimon.common.options import Options
+from pypaimon.common.options.config import CatalogOptions
+
+
+class DLFLocalFileTokenLoaderTest(unittest.TestCase):
+
+ def test_load_token_from_configured_path(self):
+ token = DLFToken(
+ access_key_id="access-key-id",
+ access_key_secret="access-key-secret",
+ security_token="security-token",
+ expiration="2099-12-01T12:00:00Z",
+ )
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ token_path = Path(temp_dir) / "token.json"
+ token_path.write_text(JSON.to_json(token), encoding="utf-8")
+ options = Options({CatalogOptions.DLF_TOKEN_PATH.key():
str(token_path)})
+
+ loader = DLFTokenLoaderFactory.create_token_loader(options)
+ loaded_token = loader.load_token()
+
+ self.assertIsInstance(loader, DLFLocalFileTokenLoader)
+ self.assertEqual(str(token_path), loader.description())
+ self.assertEqual(token.access_key_id, loaded_token.access_key_id)
+ self.assertEqual(token.access_key_secret,
loaded_token.access_key_secret)
+ self.assertEqual(token.security_token, loaded_token.security_token)
+ self.assertEqual(token.expiration, loaded_token.expiration)
+
+ def test_auth_provider_uses_token_path_without_explicit_loader(self):
+ token = DLFToken("access-key-id", "access-key-secret",
"security-token")
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ token_path = Path(temp_dir) / "token.json"
+ token_path.write_text(JSON.to_json(token), encoding="utf-8")
+ options = Options({
+ CatalogOptions.URI.key():
+ "https://cn-hangzhou-vpc.dlf.aliyuncs.com";,
+ CatalogOptions.TOKEN_PROVIDER.key(): "dlf",
+ CatalogOptions.DLF_TOKEN_PATH.key(): str(token_path),
+ })
+
+ provider = AuthProviderFactory.create_auth_provider(options)
+
+ self.assertIsInstance(provider.token_loader,
DLFLocalFileTokenLoader)
+ self.assertEqual("access-key-id",
provider.get_token().access_key_id)
+
+ def test_token_loader_takes_precedence_over_static_credentials(self):
+ file_token = DLFToken("file-ak", "file-sk", "file-sts")
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ token_path = Path(temp_dir) / "token.json"
+ token_path.write_text(JSON.to_json(file_token), encoding="utf-8")
+
+ for loader_name in (None, "local_file"):
+ with self.subTest(loader_name=loader_name):
+ option_values = {
+ CatalogOptions.URI.key():
+ "https://cn-hangzhou-vpc.dlf.aliyuncs.com";,
+ CatalogOptions.TOKEN_PROVIDER.key(): "dlf",
+ CatalogOptions.DLF_TOKEN_PATH.key(): str(token_path),
+ CatalogOptions.DLF_ACCESS_KEY_ID.key(): "static-ak",
+ CatalogOptions.DLF_ACCESS_KEY_SECRET.key():
"static-sk",
+ }
+ if loader_name is not None:
+ option_values[
+ CatalogOptions.DLF_TOKEN_LOADER.key()
+ ] = loader_name
+
+ provider = AuthProviderFactory.create_auth_provider(
+ Options(option_values)
+ )
+
+ self.assertEqual("file-ak",
provider.get_token().access_key_id)
+
+ def
test_unknown_token_loader_does_not_fallback_to_static_credentials(self):
+ options = Options({
+ CatalogOptions.URI.key():
+ "https://cn-hangzhou-vpc.dlf.aliyuncs.com";,
+ CatalogOptions.TOKEN_PROVIDER.key(): "dlf",
+ CatalogOptions.DLF_TOKEN_LOADER.key(): "unknown",
+ CatalogOptions.DLF_ACCESS_KEY_ID.key(): "static-ak",
+ CatalogOptions.DLF_ACCESS_KEY_SECRET.key(): "static-sk",
+ })
+
+ with self.assertRaisesRegex(ValueError, "Unknown DLF token loader:
unknown"):
+ AuthProviderFactory.create_auth_provider(options)
+
+ def test_loader_reads_rotated_token(self):
+ first_token = DLFToken("first-ak", "first-sk", "first-sts")
+ second_token = DLFToken("second-ak", "second-sk", "second-sts")
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ token_path = Path(temp_dir) / "token.json"
+ token_path.write_text(JSON.to_json(first_token), encoding="utf-8")
+ loader = DLFLocalFileTokenLoader(str(token_path))
+
+ self.assertEqual("first-ak", loader.load_token().access_key_id)
+ token_path.write_text(JSON.to_json(second_token), encoding="utf-8")
+ self.assertEqual("second-ak", loader.load_token().access_key_id)
+
+ def test_auth_provider_reloads_expiring_token_from_path(self):
+ expired_token = DLFToken(
+ "first-ak", "first-sk", "first-sts", "2000-01-01T00:00:00Z"
+ )
+ fresh_token = DLFToken(
+ "second-ak", "second-sk", "second-sts", "2099-01-01T00:00:00Z"
+ )
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ token_path = Path(temp_dir) / "token.json"
+ token_path.write_text(JSON.to_json(expired_token),
encoding="utf-8")
+ options = Options({
+ CatalogOptions.URI.key():
+ "https://cn-hangzhou-vpc.dlf.aliyuncs.com";,
+ CatalogOptions.TOKEN_PROVIDER.key(): "dlf",
+ CatalogOptions.DLF_TOKEN_PATH.key(): str(token_path),
+ })
+ provider = AuthProviderFactory.create_auth_provider(options)
+
+ self.assertEqual("first-ak", provider.get_token().access_key_id)
+ token_path.write_text(JSON.to_json(fresh_token), encoding="utf-8")
+ self.assertEqual("second-ak", provider.get_token().access_key_id)
+
+ def test_malformed_token_file_does_not_leak_credentials(self):
+ secret = "STSSECRET_AKID_9999"
+
+ with tempfile.TemporaryDirectory() as temp_dir:
+ token_path = Path(temp_dir) / "token.json"
+ token_path.write_text(
+ '{"AccessKeyId":"akid","AccessKeySecret":"%s" INVALID_JSON'
+ % secret,
+ encoding="utf-8",
+ )
+
+ with self.assertRaisesRegex(RuntimeError, "Failed to parse token
file") as ctx:
+ DLFLocalFileTokenLoader.read_token(str(token_path),
max_retries=1)
+
+ self.assertNotIn(secret, str(ctx.exception))
+ self.assertIsNone(ctx.exception.__cause__)
+
+
+if __name__ == "__main__":
+ unittest.main()
