From bd1621e1a4a4ae36d85ffd25fd258155ae66a8e1 Mon Sep 17 00:00:00 2001 From: moke-HU <25567926679@qq.com> Date: Thu, 13 Aug 2026 23:29:45 +0800 Subject: [PATCH 1/2] [filesystem] Fix COS STS policy resource APPID --- .../cos/token/COSSecurityTokenProvider.java | 30 +++++++++-- .../token/COSSecurityTokenProviderTest.java | 52 +++++++++++++++++++ 2 files changed, 77 insertions(+), 5 deletions(-) create mode 100644 fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProviderTest.java diff --git a/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java index 5b43908513a..3a57fb48a91 100644 --- a/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java +++ b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java @@ -132,11 +132,11 @@ private static String resolvePolicy(Configuration conf, URI fsUri, String region * Builds a default STS policy that grants {@code name/cos:*} only on the bucket (and optional * key prefix) referenced by the given fsUri. * - *

COS resource format used here: {@code qcs::cos::uid/*:/*}. The - * wildcard owner uid keeps the policy independent of the account uin while still restricting - * access to a specific bucket. + *

COS resource format used here: {@code + * qcs::cos::uid/:/*}. The APPID is the numeric suffix of a COS + * bucket name. */ - private static String buildBucketScopedPolicy(URI fsUri, String region) { + static String buildBucketScopedPolicy(URI fsUri, String region) { String bucket = fsUri.getAuthority(); if (bucket == null || bucket.isEmpty()) { // Fall back to all-resources policy if we cannot derive the bucket from fsUri. @@ -156,6 +156,17 @@ private static String buildBucketScopedPolicy(URI fsUri, String region) { + "}"; } + int appIdSeparator = bucket.lastIndexOf('-'); + String appId = appIdSeparator < 0 ? "" : bucket.substring(appIdSeparator + 1); + if (appId.isEmpty() || !appId.chars().allMatch(Character::isDigit)) { + throw new IllegalArgumentException( + "Unable to derive COS APPID from bucket " + + bucket + + ". Expected a bucket name ending in '-'; configure " + + SECURITY_TOKEN_POLICY + + " explicitly if a custom resource policy is required."); + } + String path = fsUri.getPath(); String prefix; if (path == null || path.isEmpty() || "/".equals(path)) { @@ -168,7 +179,16 @@ private static String buildBucketScopedPolicy(URI fsUri, String region) { } } - String resource = "qcs::cos:" + region + ":uid/*:" + bucket + "/" + prefix + "*"; + String resource = + "qcs::cos:" + + region + + ":uid/" + + appId + + ":" + + bucket + + "/" + + prefix + + "*"; LOG.info("Using bucket-scoped STS policy with resource: {}", resource); return "{" + "\"version\": \"2.0\"," diff --git a/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProviderTest.java b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProviderTest.java new file mode 100644 index 00000000000..d5acab36442 --- /dev/null +++ b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProviderTest.java @@ -0,0 +1,52 @@ +/* + * 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. + */ + +package org.apache.fluss.fs.cos.token; + +import org.junit.jupiter.api.Test; + +import java.net.URI; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.assertj.core.api.Assertions.assertThatThrownBy; + +class COSSecurityTokenProviderTest { + + @Test + void testBuildBucketScopedPolicyUsesBucketAppId() { + String policy = + COSSecurityTokenProvider.buildBucketScopedPolicy( + URI.create("cosn://data-test-1370497452/fluss/remote-data"), + "ap-guangzhou"); + + assertThat(policy) + .contains( + "qcs::cos:ap-guangzhou:uid/1370497452:data-test-1370497452/fluss/remote-data/*") + .doesNotContain("uid/*"); + } + + @Test + void testBuildBucketScopedPolicyRejectsBucketWithoutAppId() { + assertThatThrownBy( + () -> + COSSecurityTokenProvider.buildBucketScopedPolicy( + URI.create("cosn://bucket/fluss/remote-data"), + "ap-guangzhou")) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("Expected a bucket name ending in '-'"); + } +} From b724ce226c49dacf23f611048a66500de088120b Mon Sep 17 00:00:00 2001 From: moke-HU <25567926679@qq.com> Date: Fri, 14 Aug 2026 01:28:24 +0800 Subject: [PATCH 2/2] [filesystem] Preserve COS session credentials --- fluss-filesystems/fluss-fs-cos/pom.xml | 2 +- .../cos/token/COSSecurityTokenProvider.java | 15 +-- .../cos/token/COSSessionCredentialsTest.java | 106 ++++++++++++++++++ 3 files changed, 110 insertions(+), 13 deletions(-) create mode 100644 fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/token/COSSessionCredentialsTest.java diff --git a/fluss-filesystems/fluss-fs-cos/pom.xml b/fluss-filesystems/fluss-fs-cos/pom.xml index 7887ee2a111..63b90c1e13e 100644 --- a/fluss-filesystems/fluss-fs-cos/pom.xml +++ b/fluss-filesystems/fluss-fs-cos/pom.xml @@ -31,7 +31,7 @@ Fluss : FileSystems : COS FS - 3.3.5 + 3.5.0 5.6.139 3.1.678 diff --git a/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java index 3a57fb48a91..dc8e1867905 100644 --- a/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java +++ b/fluss-filesystems/fluss-fs-cos/src/main/java/org/apache/fluss/fs/cos/token/COSSecurityTokenProvider.java @@ -132,9 +132,8 @@ private static String resolvePolicy(Configuration conf, URI fsUri, String region * Builds a default STS policy that grants {@code name/cos:*} only on the bucket (and optional * key prefix) referenced by the given fsUri. * - *

COS resource format used here: {@code - * qcs::cos::uid/:/*}. The APPID is the numeric suffix of a COS - * bucket name. + *

COS resource format used here: {@code qcs::cos::uid/:/*}. + * The APPID is the numeric suffix of a COS bucket name. */ static String buildBucketScopedPolicy(URI fsUri, String region) { String bucket = fsUri.getAuthority(); @@ -180,15 +179,7 @@ static String buildBucketScopedPolicy(URI fsUri, String region) { } String resource = - "qcs::cos:" - + region - + ":uid/" - + appId - + ":" - + bucket - + "/" - + prefix - + "*"; + "qcs::cos:" + region + ":uid/" + appId + ":" + bucket + "/" + prefix + "*"; LOG.info("Using bucket-scoped STS policy with resource: {}", resource); return "{" + "\"version\": \"2.0\"," diff --git a/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/token/COSSessionCredentialsTest.java b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/token/COSSessionCredentialsTest.java new file mode 100644 index 00000000000..17ad7689ee0 --- /dev/null +++ b/fluss-filesystems/fluss-fs-cos/src/test/java/org/apache/fluss/fs/cos/token/COSSessionCredentialsTest.java @@ -0,0 +1,106 @@ +/* + * 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. + */ + +package org.apache.fluss.fs.cos.token; + +import org.apache.fluss.fs.token.Credentials; +import org.apache.fluss.fs.token.CredentialsJsonSerde; +import org.apache.fluss.fs.token.ObtainedSecurityToken; + +import com.qcloud.cos.COSClient; +import com.qcloud.cos.auth.COSCredentials; +import com.qcloud.cos.auth.COSCredentialsProvider; +import com.qcloud.cos.auth.COSSessionCredentials; +import org.apache.hadoop.conf.Configuration; +import org.apache.hadoop.fs.cosn.CosNFileSystem; +import org.junit.jupiter.api.Test; + +import java.lang.reflect.Field; +import java.lang.reflect.InvocationHandler; +import java.lang.reflect.Proxy; +import java.net.URI; +import java.util.HashMap; +import java.util.Map; + +import static org.assertj.core.api.Assertions.assertThat; + +class COSSessionCredentialsTest { + + private static final String SESSION_TOKEN = "test-session-token"; + + @Test + void testCosNFileSystemPreservesSessionToken() throws Exception { + ObtainedSecurityToken token = + new ObtainedSecurityToken( + "cosn", + CredentialsJsonSerde.toJson( + new Credentials( + "test-access-key", "test-secret-key", SESSION_TOKEN)), + null, + additionInfos()); + new COSSecurityTokenReceiver().onNewTokensObtained(token); + + Configuration configuration = new Configuration(false); + configuration.set( + "fs.cosn.credentials.provider", DynamicTemporaryCOSCredentialsProvider.NAME); + additionInfos().forEach(configuration::set); + + try (CosNFileSystem fileSystem = new CosNFileSystem()) { + fileSystem.initialize(URI.create("cosn://test-bucket-1234567890"), configuration); + + COSCredentials credentials = getCosClientCredentials(fileSystem); + assertThat(credentials).isInstanceOf(COSSessionCredentials.class); + assertThat(((COSSessionCredentials) credentials).getSessionToken()) + .isEqualTo(SESSION_TOKEN); + } + } + + private static COSCredentials getCosClientCredentials(CosNFileSystem fileSystem) + throws Exception { + Object store = getField(CosNFileSystem.class, "store").get(fileSystem); + if (Proxy.isProxyClass(store.getClass())) { + InvocationHandler handler = Proxy.getInvocationHandler(store); + Object descriptor = getField(handler.getClass(), "proxyDescriptor").get(handler); + Object proxyInfo = getField(descriptor.getClass(), "proxyInfo").get(descriptor); + store = getField(proxyInfo.getClass(), "proxy").get(proxyInfo); + } + COSClient cosClient = (COSClient) getField(store.getClass(), "cosClient").get(store); + COSCredentialsProvider provider = + (COSCredentialsProvider) getField(COSClient.class, "credProvider").get(cosClient); + return provider.getCredentials(); + } + + private static Field getField(Class type, String name) throws Exception { + for (Class current = type; current != null; current = current.getSuperclass()) { + try { + Field field = current.getDeclaredField(name); + field.setAccessible(true); + return field; + } catch (NoSuchFieldException ignored) { + // Continue with the superclass. + } + } + throw new NoSuchFieldException(type.getName() + "." + name); + } + + private static Map additionInfos() { + Map additionInfos = new HashMap<>(); + additionInfos.put("fs.cosn.userinfo.region", "ap-guangzhou"); + additionInfos.put("fs.cosn.bucket.endpoint_suffix", "cos.ap-guangzhou.myqcloud.com"); + return additionInfos; + } +}