JingsongLi commented on code in PR #8460:
URL: https://github.com/apache/paimon/pull/8460#discussion_r3525095730
##########
paimon-filesystems/paimon-oss-impl/src/main/java/org/apache/paimon/oss/OSSFileIO.java:
##########
@@ -273,112 +280,205 @@ private static OSSClient
getOssClient(AliyunOSSFileSystem fs) throws Exception {
return ReflectionUtils.getPrivateFieldValue(store, "ossClient");
}
- /**
- * Enable SSE-KMS with a customer CMK. hadoop-aliyun exposes no CMK key-id
option, so swap the
- * OSS client's object/multipart operations for subclasses that stamp the
OSS-native SSE-KMS
- * headers on the write paths: PutObject, server-side CopyObject
(rename/commit) and
- * InitiateMultipartUpload. UploadPart/HeadObject/GetObject are left
untouched (stamping SSE
- * headers on those makes OSS reject the request with 400).
- */
- private void enableSseKms(AliyunOSSFileSystem fs, String kmsKeyId) {
+ /** The Paimon SSE config from the three keys, or null when none is set. */
+ private SseConfig configuredSse() {
+ return resolveSse(
+ hadoopOptions.get(OSS_SSE_METHOD),
+ hadoopOptions.get(OSS_SSE_KMS_KEY_ID),
+ hadoopOptions.get(OSS_SSE_DATA_ENCRYPTION));
+ }
+
+ /** Parse the three SSE keys into an {@link SseConfig}, or null. Throws on
bad input. */
+ static SseConfig resolveSse(String method, String keyId, String
dataEncryption) {
+ // Reject present-but-blank values fail-fast instead of silently
writing plaintext.
+ checkNotBlankIfSet(method, OSS_SSE_METHOD);
+ checkNotBlankIfSet(keyId, OSS_SSE_KMS_KEY_ID);
+ checkNotBlankIfSet(dataEncryption, OSS_SSE_DATA_ENCRYPTION);
+ method = method == null ? null : method.trim();
+ keyId = keyId == null ? null : keyId.trim();
+ dataEncryption = dataEncryption == null ? null : dataEncryption.trim();
+ if (method == null && keyId == null && dataEncryption == null) {
+ return null;
+ }
+ // A key id / data-encryption implies KMS; default the method to KMS
when unset.
+ method = canonicalSseMethod(method == null ? SSE_METHOD_KMS : method);
+ if (keyId != null) {
+ checkArgument(
+ keyId.chars().noneMatch(Character::isWhitespace),
+ "Invalid value for '%s': the CMK key id must not contain
whitespace/newlines.",
+ OSS_SSE_KMS_KEY_ID);
+ checkArgument(
+ SSE_METHOD_KMS.equals(method),
+ "'%s' requires '%s=KMS', but got '%s'.",
+ OSS_SSE_KMS_KEY_ID,
+ OSS_SSE_METHOD,
+ method);
+ }
+ if (dataEncryption != null) {
+ checkArgument(
+ SSE_METHOD_KMS.equals(method),
+ "'%s' requires '%s=KMS', but got '%s'.",
+ OSS_SSE_DATA_ENCRYPTION,
+ OSS_SSE_METHOD,
+ method);
+ checkArgument(
+ SSE_DATA_SM4.equalsIgnoreCase(dataEncryption),
+ "'%s' only supports 'SM4', but got '%s'.",
+ OSS_SSE_DATA_ENCRYPTION,
+ dataEncryption);
+ dataEncryption = SSE_DATA_SM4;
+ }
+ return new SseConfig(method, keyId, dataEncryption);
+ }
+
+ /** Reject a config value that is present but blank (encryption requested
yet misconfigured). */
+ private static void checkNotBlankIfSet(String value, String key) {
+ checkArgument(
+ value == null || !StringUtils.isNullOrWhitespaceOnly(value),
+ "'%s' is set but blank.",
+ key);
+ }
+
+ /** Canonicalize the SSE method to the exact OSS header value; reject
unknown values. */
+ private static String canonicalSseMethod(String method) {
+ if (SSE_METHOD_AES256.equalsIgnoreCase(method)) {
+ return SSE_METHOD_AES256;
+ }
+ if (SSE_METHOD_KMS.equalsIgnoreCase(method)) {
+ return SSE_METHOD_KMS;
+ }
+ if (SSE_DATA_SM4.equalsIgnoreCase(method)) {
+ return SSE_DATA_SM4;
+ }
+ throw new IllegalArgumentException(
+ "'"
+ + OSS_SSE_METHOD
+ + "' must be one of AES256/KMS/SM4, but got '"
+ + method
+ + "'.");
+ }
+
+ /** Resolved OSS server-side-encryption settings. */
+ static final class SseConfig {
+ final String method;
+ final String keyId;
+ final String dataEnc;
+
+ SseConfig(String method, String keyId, String dataEnc) {
+ this.method = method;
+ this.keyId = keyId;
+ this.dataEnc = dataEnc;
+ }
+ }
+
+ /** Swap in the SSE-stamping object/multipart operations (write paths
only). */
+ private void enableSse(AliyunOSSFileSystem fs, SseConfig sse) {
try {
- swapSseKmsOperations(getOssClient(fs), kmsKeyId);
+ swapSseOperations(getOssClient(fs), sse);
} catch (Exception e) {
- LOG.error("Failed to enable SSE-KMS BYOK.", e);
- throw new RuntimeException("Failed to enable SSE-KMS BYOK.", e);
+ LOG.error("Failed to enable OSS server-side encryption.", e);
+ throw new RuntimeException("Failed to enable OSS server-side
encryption.", e);
}
}
- /**
- * Swap the OSS client's object/multipart operations for the CMK-stamping
subclasses. Package
- * private so a unit test can pin the reflected field names and getters
against the bundled
- * aliyun-oss-sdk version.
- */
- static void swapSseKmsOperations(OSSClient ossClient, String kmsKeyId)
throws Exception {
+ /** Swap in the SSE-stamping operations; package-private so a test pins
reflected names. */
+ static void swapSseOperations(OSSClient ossClient, SseConfig sse) throws
Exception {
ServiceClient serviceClient =
ReflectionUtils.getPrivateFieldValue(ossClient,
"serviceClient");
CredentialsProvider credsProvider =
ReflectionUtils.getPrivateFieldValue(ossClient,
"credsProvider");
// Replacement operations must inherit the endpoint or requests NPE.
URI endpoint = ossClient.getEndpoint();
- SseKmsObjectOperation objectOperation =
- new SseKmsObjectOperation(serviceClient, credsProvider,
kmsKeyId);
+ SseObjectOperation objectOperation =
+ new SseObjectOperation(serviceClient, credsProvider, sse);
objectOperation.setEndpoint(endpoint);
- SseKmsMultipartOperation multipartOperation =
- new SseKmsMultipartOperation(serviceClient, credsProvider,
kmsKeyId);
+ SseMultipartOperation multipartOperation =
+ new SseMultipartOperation(serviceClient, credsProvider, sse);
multipartOperation.setEndpoint(endpoint);
ReflectionUtils.setPrivateFieldValue(ossClient, "objectOperation",
objectOperation);
ReflectionUtils.setPrivateFieldValue(ossClient, "multipartOperation",
multipartOperation);
- // Fail closed: if a future SDK change makes the swap a no-op, fail
loudly
- // rather than silently writing objects without the customer CMK.
+ // Fail closed: if the swap didn't take effect, refuse to write
unencrypted.
if (ossClient.getObjectOperation() != objectOperation
|| ossClient.getMultipartOperation() != multipartOperation) {
throw new IllegalStateException(
- "SSE-KMS BYOK operation swap did not take effect; refusing
to write "
- + "without server-side encryption. The
aliyun-oss-sdk internals "
- + "may have changed.");
+ "OSS SSE operation swap did not take effect; refusing to
write without "
+ + "server-side encryption. The aliyun-oss-sdk
internals may have changed.");
}
}
- /** Stamps the customer CMK on the given metadata, allocating one if
absent. */
- private static ObjectMetadata applySseKms(ObjectMetadata metadata, String
kmsKeyId) {
+ /** Stamp the SSE headers on object metadata (PutObject / multipart init).
*/
+ static ObjectMetadata applySse(ObjectMetadata metadata, SseConfig sse) {
if (metadata == null) {
metadata = new ObjectMetadata();
}
String existing = metadata.getServerSideEncryption();
- if (existing != null && !SSE_KMS_ALGORITHM.equals(existing)) {
+ if (existing != null && !sse.method.equals(existing)) {
LOG.warn(
- "Customer CMK SSE-KMS is overriding the previously-set
server-side "
- + "encryption algorithm '{}' with 'KMS'.",
- existing);
+ "Paimon SSE is overriding the previously-set server-side
encryption "
+ + "method '{}' with '{}'.",
+ existing,
+ sse.method);
+ }
+ metadata.setServerSideEncryption(sse.method);
+ if (sse.keyId != null) {
+ metadata.setServerSideEncryptionKeyId(sse.keyId);
+ }
+ if (sse.dataEnc != null) {
+ metadata.setServerSideDataEncryption(sse.dataEnc);
}
- metadata.setServerSideEncryption(SSE_KMS_ALGORITHM);
- metadata.setServerSideEncryptionKeyId(kmsKeyId);
return metadata;
}
- /** Stamps the customer CMK on every simple PutObject. */
- static class SseKmsObjectOperation extends OSSObjectOperation {
- private final String kmsKeyId;
+ /** Stamp the SSE headers on a server-side CopyObject request
(rename/commit). */
+ static CopyObjectRequest applySse(CopyObjectRequest request, SseConfig
sse) {
+ // Use request fields, not newObjectMetadata (forces REPLACE, drops
source metadata).
+ // No data-encryption field; addHeader works since doOperation merges
headers pre-sign.
+ request.setServerSideEncryption(sse.method);
+ if (sse.keyId != null) {
+ request.setServerSideEncryptionKeyId(sse.keyId);
+ }
+ if (sse.dataEnc != null) {
+ request.addHeader(OSSHeaders.OSS_SERVER_SIDE_DATA_ENCRYPTION,
sse.dataEnc);
Review Comment:
This does not seem to reach the actual CopyObject HTTP request. In
aliyun-sdk-oss 3.13.2, `OSSObjectOperation.copyObject()` creates a fresh local
headers map and populates it only through
`populateCopyObjectHeaders(copyObjectRequest, headers)`, which reads the
dedicated CopyObjectRequest fields for SSE method/key-id but does not merge
`WebServiceRequest.getHeaders()`. So `fs.oss.server-side-data-encryption=SM4`
will be recorded on the request object but dropped before signing/sending
CopyObject, meaning rename/commit copies are not stamped as documented. The
current unit test only checks `kms.getHeaders()`, so it misses this SDK
request-building path. We probably need to override/build the CopyObject
request headers directly, or use an SDK API that has a dedicated
data-encryption field, and add a test that covers the final headers passed to
the operation.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
