plusplusjiajia commented on code in PR #8460:
URL: https://github.com/apache/paimon/pull/8460#discussion_r3525117244
##########
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:
@JingsongLi Thanks for digging into the SDK internals! You're right that
populateCopyObjectHeaders doesn't copy generic headers — but
OSSOperation#doOperation merges them before signing (verified on the bundled
3.13.2): request.getHeaders().putAll(originalRequest.getHeaders());
copyObject passes the CopyObjectRequest as originalRequest, and addHeader
writes to that same map, so the header is signed and sent.
Added testCopyObjectSendsSseHeadersOnWire: it runs the real copyObject
request-building against a capturing ServiceClient and asserts the final
request carries all three SSE headers. Also verified against live OSS:
CopyObject with KMS+SM4 returns 200 and HEAD of the destination shows
x-oss-server-side-data-encryption: SM4.
--
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]
