This is an automated email from the ASF dual-hosted git repository. damccorm pushed a commit to branch users/damccorm/secretEncoding in repository https://gitbox.apache.org/repos/asf/beam.git
commit c651a9c1231de26b054cc9ef0223d493e7496f26 Author: Danny Mccormick <[email protected]> AuthorDate: Tue Oct 7 20:12:29 2025 -0400 Use consistent encoding for GBEK across languages --- .../java/org/apache/beam/sdk/options/PipelineOptions.java | 11 ++++++++--- .../org/apache/beam/sdk/transforms/GroupByEncryptedKey.java | 10 ++++++---- .../apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java | 7 +++++-- .../java/org/apache/beam/sdk/transforms/GroupByKeyIT.java | 6 +++++- .../java/org/apache/beam/sdk/transforms/GroupByKeyTest.java | 5 ++++- sdks/python/apache_beam/options/pipeline_options.py | 8 ++++++-- 6 files changed, 34 insertions(+), 13 deletions(-) diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java index 62022b219c2..989e3a1e319 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/options/PipelineOptions.java @@ -420,7 +420,7 @@ public interface PipelineOptions extends HasDisplayData { * <p>Beam will infer the secret type and value based on the secret itself. This guarantees that * any data at rest during the performing a GBK, so this can be used to guarantee that data is not * unencrypted. Runners with this behavior include the Dataflow, Flink, and Spark runners. The - * option should be structured like: + * secret should be a url safe base64 encoded 32 byte value. The option should be structured like: * * <pre><code> * --gbek=type:<secret_type>;<secret_param>:<value> @@ -432,14 +432,19 @@ public interface PipelineOptions extends HasDisplayData { * --gbek=type:GcpSecret;version_name:my_secret/versions/latest" * </code></pre> * - * All variables should use snake case to allow consistency across languages. + * All variables should use snake case to allow consistency across languages. For an example of + * generating a properly formatted secret, see + * https://github.com/apache/beam/blob/c8df4da229da49d533491857e1bb4ab5dbf4fd37/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java#L82 */ @Description( "When set, will replace all GroupByKey transforms in the pipeline the option. Beam will" + " infer the secret type and value based on the secret itself. This guarantees that" + " any data at rest during the performing a GBK, so this can be used to guarantee" + " that data is not unencrypted. Runners with this behavior include the Dataflow," - + " Flink, and Spark runners. The option should be structured like:" + + " Flink, and Spark runners. The secret should be a url safe base64 encoded 32 byte" + + " value. For an example of generating a properly formatted secret, see" + + " https://github.com/apache/beam/blob/c8df4da229da49d533491857e1bb4ab5dbf4fd37/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java#L82"; + + " When passing in the gbek option, it should be structured like:" + " --gbek=type:<secret_type>;<secret_param>:<value>, for example " + " --gbek=type:GcpSecret;version_name:my_secret/versions/latest. All variables " + " should use snake case to allow consistency across languages.") diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java index 6ed0a31b3b9..1f4b7535d89 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java +++ b/sdks/java/core/src/main/java/org/apache/beam/sdk/transforms/GroupByEncryptedKey.java @@ -39,8 +39,8 @@ import org.apache.beam.sdk.values.PCollection; * the output. This is useful when the keys contain sensitive data that should not be stored at rest * by the runner. * - * <p>The transform requires a {@link Secret} which returns a 32 byte secret which can be used to - * generate a {@link SecretKeySpec} object using the HmacSHA256 algorithm. + * <p>The transform requires a {@link Secret} which returns a base64 encoded 32 byte secret which + * can be used to generate a {@link SecretKeySpec} object using the HmacSHA256 algorithm. * * <p>Note the following caveats: 1) Runners can implement arbitrary materialization steps, so this * does not guarantee that the whole pipeline will not have unencrypted data at rest by itself. 2) @@ -153,7 +153,7 @@ public class GroupByEncryptedKey<K, V> @Setup public void setup() { try { - byte[] secretBytes = this.hmacKey.getSecretBytes(); + byte[] secretBytes = java.util.Base64.getUrlDecoder().decode(this.hmacKey.getSecretBytes()); this.mac = Mac.getInstance("HmacSHA256"); this.mac.init(new SecretKeySpec(secretBytes, "HmacSHA256")); this.cipher = Cipher.getInstance("AES/GCM/NoPadding"); @@ -229,7 +229,9 @@ public class GroupByEncryptedKey<K, V> public void setup() { try { this.cipher = Cipher.getInstance("AES/GCM/NoPadding"); - this.secretKeySpec = new SecretKeySpec(this.hmacKey.getSecretBytes(), "AES"); + this.secretKeySpec = + new SecretKeySpec( + java.util.Base64.getUrlDecoder().decode(this.hmacKey.getSecretBytes()), "AES"); } catch (Exception ex) { throw new RuntimeException( "Failed to initialize cryptography libraries needed for GroupByEncryptedKey", ex); diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java index ba4c50e5a41..876483a7ddb 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByEncryptedKeyTest.java @@ -58,7 +58,7 @@ public class GroupByEncryptedKeyTest implements Serializable { private static class FakeSecret implements Secret { private final byte[] secret = - "aKwI2PmqYFt2p5tNKCyBS5qYmHhHsGZc".getBytes(Charset.defaultCharset()); + "YUt3STJQbXFZRnQycDV0TktDeUJTNXFZV0hoSHNHWmM".getBytes(Charset.defaultCharset()); @Override public byte[] getSecretBytes() { @@ -123,7 +123,10 @@ public class GroupByEncryptedKeyTest implements Serializable { byte[] secretBytes = new byte[32]; new SecureRandom().nextBytes(secretBytes); client.addSecretVersion( - secretName, SecretPayload.newBuilder().setData(ByteString.copyFrom(secretBytes)).build()); + secretName, + SecretPayload.newBuilder() + .setData( + ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes))) } gcpSecret = new GcpSecret(secretName.toString() + "/versions/latest"); } diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java index 60477a4c242..3dd9975463b 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java @@ -82,7 +82,11 @@ public class GroupByKeyIT { byte[] secretBytes = new byte[32]; new SecureRandom().nextBytes(secretBytes); client.addSecretVersion( - secretName, SecretPayload.newBuilder().setData(ByteString.copyFrom(secretBytes)).build()); + secretName, + SecretPayload.newBuilder() + .setData( + ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes))) + .build()); } gcpSecretVersionName = secretName.toString() + "/versions/latest"; } diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java index 326da99f1a8..c6e84fedf85 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyTest.java @@ -153,7 +153,10 @@ public class GroupByKeyTest implements Serializable { new SecureRandom().nextBytes(secretBytes); client.addSecretVersion( secretName, - SecretPayload.newBuilder().setData(ByteString.copyFrom(secretBytes)).build()); + SecretPayload.newBuilder() + .setData( + ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes))) + .build()); } gcpSecretVersionName = secretName.toString() + "/versions/latest"; } diff --git a/sdks/python/apache_beam/options/pipeline_options.py b/sdks/python/apache_beam/options/pipeline_options.py index 2d3b8b49d8d..d4903cfb009 100644 --- a/sdks/python/apache_beam/options/pipeline_options.py +++ b/sdks/python/apache_beam/options/pipeline_options.py @@ -1726,8 +1726,12 @@ class SetupOptions(PipelineOptions): 'secret itself. This guarantees that any data at rest during the ' 'GBK will be encrypted. Many runners only store data at rest when ' 'performing a GBK, so this can be used to guarantee that data is ' - 'not unencrypted. Runners with this behavior include the ' - 'Dataflow, Flink, and Spark runners. The option should be ' + 'not unencrypted. The secret should be a url safe base64 encoded ' + '32 byte value. To generate a secret in this format, you can use ' + 'Secret.generate_secret_bytes(). For an example of this, see ' + 'https://github.com/apache/beam/blob/c8df4da229da49d533491857e1bb4ab5dbf4fd37/sdks/python/apache_beam/transforms/util_test.py#L356. ' + 'Runners with this behavior include the Dataflow, ' + 'Flink, and Spark runners. The option should be ' 'structured like: ' '--gbek=type:<secret_type>;<secret_param>:<value>, for example ' '--gbek=type:GcpSecret;version_name:my_secret/versions/latest'))
