diff --git a/.github/trigger_files/beam_PostCommit_XVR_Direct.json b/.github/trigger_files/beam_PostCommit_XVR_Direct.json index 3b864120bf95..c04f8e34f87f 100644 --- a/.github/trigger_files/beam_PostCommit_XVR_Direct.json +++ b/.github/trigger_files/beam_PostCommit_XVR_Direct.json @@ -1,4 +1,4 @@ { - "modification": 1, + "modification": 2, "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner" } diff --git a/.github/trigger_files/beam_PostCommit_XVR_Flink.json b/.github/trigger_files/beam_PostCommit_XVR_Flink.json index a60830fe7d2a..57f49e9baefe 100644 --- a/.github/trigger_files/beam_PostCommit_XVR_Flink.json +++ b/.github/trigger_files/beam_PostCommit_XVR_Flink.json @@ -1,5 +1,4 @@ { - "modification": 3, - "trigger-2026-04-04": "portable_runner expand_sdf opt-in", - "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner" + "modification": 4, + "trigger-2026-04-04": "portable_runner expand_sdf opt-in" } diff --git a/.github/trigger_files/beam_PostCommit_XVR_JavaUsingPython_Dataflow.json b/.github/trigger_files/beam_PostCommit_XVR_JavaUsingPython_Dataflow.json index 0ca37f8c8e27..7ab7bcd9a9c6 100644 --- a/.github/trigger_files/beam_PostCommit_XVR_JavaUsingPython_Dataflow.json +++ b/.github/trigger_files/beam_PostCommit_XVR_JavaUsingPython_Dataflow.json @@ -1,5 +1,4 @@ { - "comment": "Modify this file in a trivial way to cause this test suite to run.", - "modification": 1, - "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner" + "comment": "Modify this file in a trivial way to cause this test suite to run.", + "modification": 2 } diff --git a/.github/trigger_files/beam_PostCommit_XVR_Spark3.json b/.github/trigger_files/beam_PostCommit_XVR_Spark3.json index 74a5d652c7b9..4fad3843d36c 100644 --- a/.github/trigger_files/beam_PostCommit_XVR_Spark3.json +++ b/.github/trigger_files/beam_PostCommit_XVR_Spark3.json @@ -1,4 +1,4 @@ { - "trigger-2026-07-08": "portable_runner expand_sdf opt-in 2", - "https://github.com/apache/beam/pull/39990": "removing dead code from FnApiDoFnRunner" + "modification": 1, + "trigger-2026-07-08": "portable_runner expand_sdf opt-in 2" } diff --git a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy index da54354f01bd..be10dd4779ca 100644 --- a/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy +++ b/buildSrc/src/main/groovy/org/apache/beam/gradle/BeamModulePlugin.groovy @@ -803,6 +803,7 @@ class BeamModulePlugin implements Plugin { grpc_google_cloud_pubsub_v1 : "com.google.api.grpc:grpc-google-cloud-pubsub-v1", // google_cloud_platform_libraries_bom sets version grpc_google_common_protos : "com.google.api.grpc:grpc-google-common-protos", // google_cloud_platform_libraries_bom sets version grpc_grpclb : "io.grpc:grpc-grpclb", // google_cloud_platform_libraries_bom sets version + grpc_inprocess : "io.grpc:grpc-inprocess", // google_cloud_platform_libraries_bom sets version grpc_protobuf : "io.grpc:grpc-protobuf", // google_cloud_platform_libraries_bom sets version grpc_protobuf_lite : "io.grpc:grpc-protobuf-lite", // google_cloud_platform_libraries_bom sets version grpc_netty : "io.grpc:grpc-netty", // google_cloud_platform_libraries_bom sets version diff --git a/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml b/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml index 76dffb30271e..833fd30cf218 100644 --- a/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml +++ b/sdks/java/build-tools/src/main/resources/beam/checkstyle/suppressions.xml @@ -58,11 +58,9 @@ - - - - - + + + diff --git a/sdks/java/core/build.gradle b/sdks/java/core/build.gradle index f532e9d14166..f95a8a53906b 100644 --- a/sdks/java/core/build.gradle +++ b/sdks/java/core/build.gradle @@ -102,15 +102,6 @@ dependencies { shadow library.java.slf4j_api shadow library.java.snappy_java shadow library.java.joda_time - implementation enforcedPlatform(library.java.google_cloud_platform_libraries_bom) - implementation library.java.gax - implementation library.java.google_cloud_kms - implementation library.java.proto_google_cloud_kms_v1 - implementation library.java.google_cloud_tink - implementation library.java.google_cloud_secret_manager - implementation library.java.proto_google_cloud_secret_manager_v1 - implementation library.java.protobuf_java - permitUnusedDeclared enforcedPlatform(library.java.google_cloud_platform_libraries_bom) provided library.java.json_org implementation library.java.everit_json_schema implementation library.java.guava @@ -133,10 +124,6 @@ dependencies { shadowTest library.java.log4j shadowTest library.java.log4j2_api shadowTest library.java.jamm - shadowTest enforcedPlatform(library.java.google_cloud_platform_libraries_bom) - shadowTest library.java.google_cloud_secret_manager - shadowTest library.java.google_cloud_kms - shadowTest library.java.google_cloud_tink testRuntimeOnly library.java.slf4j_jdk14 } 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 a929d62d29d7..7bcebcef5b7e 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 @@ -17,18 +17,11 @@ */ package org.apache.beam.sdk.transforms; -import static org.junit.Assert.assertThrows; - -import com.google.cloud.secretmanager.v1.ProjectName; -import com.google.cloud.secretmanager.v1.SecretManagerServiceClient; -import com.google.cloud.secretmanager.v1.SecretName; -import com.google.cloud.secretmanager.v1.SecretPayload; -import com.google.protobuf.ByteString; -import java.io.IOException; import java.io.Serializable; import java.nio.charset.Charset; -import java.security.SecureRandom; +import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.Base64; import java.util.List; import java.util.stream.Collectors; import java.util.stream.StreamSupport; @@ -39,14 +32,11 @@ import org.apache.beam.sdk.testing.NeedsRunner; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; -import org.apache.beam.sdk.util.GcpHsmGeneratedSecret; -import org.apache.beam.sdk.util.GcpSecret; +import org.apache.beam.sdk.util.RawSecret; import org.apache.beam.sdk.util.Secret; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.checkerframework.checker.nullness.qual.Nullable; -import org.junit.AfterClass; -import org.junit.BeforeClass; import org.junit.Rule; import org.junit.Test; import org.junit.experimental.categories.Category; @@ -100,138 +90,9 @@ public void testGroupByKeyFakeSecret() { p.run(); } - private static final String PROJECT_ID = "apache-beam-testing"; - private static final String SECRET_ID = "gbek-test"; - private static Secret gcpSecret; - private static Secret gcpHsmGeneratedSecret; - private static final String KEY_RING_ID = "gbek-test-key-ring"; - private static final String KEY_ID = "gbek-test-key"; - - @BeforeClass - public static void setup() throws IOException { - SecretManagerServiceClient client = SecretManagerServiceClient.create(); - ProjectName projectName = ProjectName.of(PROJECT_ID); - SecretName secretName = SecretName.of(PROJECT_ID, SECRET_ID); - - try { - client.getSecret(secretName); - } catch (Exception e) { - com.google.cloud.secretmanager.v1.Secret secret = - com.google.cloud.secretmanager.v1.Secret.newBuilder() - .setReplication( - com.google.cloud.secretmanager.v1.Replication.newBuilder() - .setAutomatic( - com.google.cloud.secretmanager.v1.Replication.Automatic.newBuilder() - .build()) - .build()) - .build(); - client.createSecret(projectName, SECRET_ID, secret); - byte[] secretBytes = new byte[32]; - new SecureRandom().nextBytes(secretBytes); - client.addSecretVersion( - secretName, - SecretPayload.newBuilder() - .setData(ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes))) - .build()); - } - gcpSecret = new GcpSecret(secretName.toString() + "/versions/latest"); - - try { - com.google.cloud.kms.v1.KeyManagementServiceClient kmsClient = - com.google.cloud.kms.v1.KeyManagementServiceClient.create(); - String locationId = "global"; - com.google.cloud.kms.v1.KeyRingName keyRingName = - com.google.cloud.kms.v1.KeyRingName.of(PROJECT_ID, locationId, KEY_RING_ID); - com.google.cloud.kms.v1.LocationName locationName = - com.google.cloud.kms.v1.LocationName.of(PROJECT_ID, locationId); - try { - kmsClient.getKeyRing(keyRingName); - } catch (Exception e) { - kmsClient.createKeyRing( - locationName, KEY_RING_ID, com.google.cloud.kms.v1.KeyRing.newBuilder().build()); - } - - com.google.cloud.kms.v1.CryptoKeyName keyName = - com.google.cloud.kms.v1.CryptoKeyName.of(PROJECT_ID, locationId, KEY_RING_ID, KEY_ID); - try { - kmsClient.getCryptoKey(keyName); - } catch (Exception e) { - com.google.cloud.kms.v1.CryptoKey key = - com.google.cloud.kms.v1.CryptoKey.newBuilder() - .setPurpose(com.google.cloud.kms.v1.CryptoKey.CryptoKeyPurpose.ENCRYPT_DECRYPT) - .build(); - kmsClient.createCryptoKey(keyRingName, KEY_ID, key); - } - gcpHsmGeneratedSecret = - new GcpHsmGeneratedSecret( - PROJECT_ID, - locationId, - KEY_RING_ID, - KEY_ID, - String.format("gbek-test-job-%d", new SecureRandom().nextInt(10000))); - // Validate we have crypto permissions or skip these tests. - gcpHsmGeneratedSecret.getSecretBytes(); - } catch (Exception e) { - gcpHsmGeneratedSecret = null; - } - } - - @AfterClass - public static void tearDown() throws IOException { - SecretManagerServiceClient client = SecretManagerServiceClient.create(); - SecretName secretName = SecretName.of(PROJECT_ID, SECRET_ID); - client.deleteSecret(secretName); - } - - @Test - @Category(NeedsRunner.class) - public void testGroupByKeyGcpSecret() { - List> ungroupedPairs = - Arrays.asList( - KV.of(null, 3), - KV.of("k1", 3), - KV.of("k5", Integer.MAX_VALUE), - KV.of("k5", Integer.MIN_VALUE), - KV.of("k2", 66), - KV.of("k1", 4), - KV.of(null, 5), - KV.of("k2", -33), - KV.of("k3", 0)); - - PCollection> input = - p.apply( - Create.of(ungroupedPairs) - .withCoder(KvCoder.of(NullableCoder.of(StringUtf8Coder.of()), VarIntCoder.of()))); - - PCollection>> output = - input.apply(GroupByEncryptedKey.create(gcpSecret)); - - PAssert.that(output.apply("Sort", MapElements.via(new SortValues()))) - .containsInAnyOrder( - KV.of("k1", Arrays.asList(3, 4)), - KV.of(null, Arrays.asList(3, 5)), - KV.of("k5", Arrays.asList(Integer.MIN_VALUE, Integer.MAX_VALUE)), - KV.of("k2", Arrays.asList(-33, 66)), - KV.of("k3", Arrays.asList(0))); - - p.run(); - } - - @Test - @Category(NeedsRunner.class) - public void testGroupByKeyGcpSecretThrows() { - Secret gcpSecret = new GcpSecret("bad_path/versions/latest"); - p.apply(Create.of(KV.of("k1", 1))) - .apply(GroupByEncryptedKey.create(gcpSecret)); - assertThrows(RuntimeException.class, () -> p.run()); - } - @Test @Category(NeedsRunner.class) - public void testGroupByKeyGcpHsmGeneratedSecret() { - if (gcpHsmGeneratedSecret == null) { - return; - } + public void testGroupByKeyRawSecret() { List> ungroupedPairs = Arrays.asList( KV.of(null, 3), @@ -249,8 +110,12 @@ public void testGroupByKeyGcpHsmGeneratedSecret() { Create.of(ungroupedPairs) .withCoder(KvCoder.of(NullableCoder.of(StringUtf8Coder.of()), VarIntCoder.of()))); + // GroupByEncryptedKey expects Secret#getSecretBytes() to return Base64-URL encoded + // bytes of a valid AES key (e.g. 32 bytes for AES-256). + byte[] secretBytes = "test-encryption-key-secret-12345".getBytes(StandardCharsets.UTF_8); + Secret rawSecret = new RawSecret(Base64.getUrlEncoder().encode(secretBytes)); PCollection>> output = - input.apply(GroupByEncryptedKey.create(gcpHsmGeneratedSecret)); + input.apply(GroupByEncryptedKey.create(rawSecret)); PAssert.that(output.apply("Sort", MapElements.via(new SortValues()))) .containsInAnyOrder( 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 3ff98d47939d..5464838ad4db 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 @@ -26,18 +26,12 @@ import static org.hamcrest.collection.IsIterableContainingInAnyOrder.containsInAnyOrder; import static org.junit.Assert.assertThrows; -import com.google.cloud.secretmanager.v1.ProjectName; -import com.google.cloud.secretmanager.v1.SecretManagerServiceClient; -import com.google.cloud.secretmanager.v1.SecretName; -import com.google.cloud.secretmanager.v1.SecretPayload; -import com.google.protobuf.ByteString; import java.io.DataInputStream; import java.io.DataOutputStream; import java.io.IOException; import java.io.InputStream; import java.io.OutputStream; import java.io.Serializable; -import java.security.SecureRandom; import java.util.ArrayList; import java.util.Arrays; import java.util.Collection; @@ -96,9 +90,7 @@ import org.hamcrest.Matcher; import org.joda.time.Duration; import org.joda.time.Instant; -import org.junit.AfterClass; import org.junit.Assert; -import org.junit.BeforeClass; import org.junit.Rule; import org.junit.Test; import org.junit.experimental.categories.Category; @@ -624,110 +616,6 @@ public void testLargeKeys100MB() throws Exception { } } - /** Tests validating GroupByKey behaviors with the gbek flag set. */ - @RunWith(JUnit4.class) - public static class GbekTests extends SharedTestBase { - private static final String PROJECT_ID = "apache-beam-testing"; - private static final String SECRET_ID = "gbek-test"; - public static String gcpSecretVersionName; - private static String secretId; - - @BeforeClass - public static void setup() throws IOException { - secretId = String.format("%s-%d", SECRET_ID, new SecureRandom().nextInt(10000)); - SecretManagerServiceClient client; - try { - client = SecretManagerServiceClient.create(); - } catch (IOException e) { - gcpSecretVersionName = null; - return; - } - ProjectName projectName = ProjectName.of(PROJECT_ID); - SecretName secretName = SecretName.of(PROJECT_ID, secretId); - - try { - client.getSecret(secretName); - } catch (Exception e) { - com.google.cloud.secretmanager.v1.Secret secret = - com.google.cloud.secretmanager.v1.Secret.newBuilder() - .setReplication( - com.google.cloud.secretmanager.v1.Replication.newBuilder() - .setAutomatic( - com.google.cloud.secretmanager.v1.Replication.Automatic.newBuilder() - .build()) - .build()) - .build(); - client.createSecret(projectName, secretId, secret); - byte[] secretBytes = new byte[32]; - new SecureRandom().nextBytes(secretBytes); - client.addSecretVersion( - secretName, - SecretPayload.newBuilder() - .setData(ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes))) - .build()); - } - gcpSecretVersionName = secretName.toString() + "/versions/latest"; - } - - @AfterClass - public static void tearDown() throws IOException { - if (gcpSecretVersionName != null) { - SecretManagerServiceClient client = SecretManagerServiceClient.create(); - SecretName secretName = SecretName.of(PROJECT_ID, secretId); - client.deleteSecret(secretName); - } - } - - @Test - @Category(NeedsRunner.class) - public void testGroupByKeyWithValidGcpSecretOption() { - if (gcpSecretVersionName == null) { - // Skip test if we couldn't set up secret manager - return; - } - List> ungroupedPairs = - Arrays.asList( - KV.of("k1", 3), - KV.of("k5", Integer.MAX_VALUE), - KV.of("k5", Integer.MIN_VALUE), - KV.of("k2", 66), - KV.of("k1", 4), - KV.of("k2", -33), - KV.of("k3", 0)); - - PCollection> input = - p.apply( - Create.of(ungroupedPairs) - .withCoder(KvCoder.of(StringUtf8Coder.of(), BigEndianIntegerCoder.of()))); - - p.getOptions().setGbek(String.format("type:gcpsecret;version_name:%s", gcpSecretVersionName)); - PCollection>> output = input.apply(GroupByKey.create()); - - SerializableFunction>>, Void> checker = - containsKvs( - kv("k1", 3, 4), - kv("k5", Integer.MIN_VALUE, Integer.MAX_VALUE), - kv("k2", 66, -33), - kv("k3", 0)); - PAssert.that(output).satisfies(checker); - PAssert.that(output).inWindow(GlobalWindow.INSTANCE).satisfies(checker); - - p.run(); - } - - @Test - @Category(NeedsRunner.class) - public void testGroupByKeyWithInvalidGcpSecretOption() { - if (gcpSecretVersionName == null) { - // Skip test if we couldn't set up secret manager - return; - } - p.getOptions().setGbek("type:gcpsecret;version_name:bad_path/versions/latest"); - p.apply(Create.of(KV.of("k1", 1))).apply(GroupByKey.create()); - assertThrows(RuntimeException.class, () -> p.run()); - } - } - /** Tests validating GroupByKey behaviors with windowing. */ @RunWith(JUnit4.class) public static class WindowTests extends SharedTestBase { diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/SecretTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/SecretTest.java index 446688035a5a..6810ea307b4d 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/SecretTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/SecretTest.java @@ -30,7 +30,6 @@ import java.util.Map; import java.util.Set; import java.util.TreeSet; -import org.apache.beam.sdk.util.common.ReflectHelpers; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; @@ -39,39 +38,6 @@ @RunWith(JUnit4.class) public class SecretTest { - @Test - public void testParseSecretOptionWithValidGcpSecret() { - String secretOption = "type:gcpsecret;version_name:my_secret/versions/latest"; - Secret secret = Secret.parseSecretOption(secretOption); - assertTrue(secret instanceof GcpSecret); - assertEquals("my_secret/versions/latest", ((GcpSecret) secret).getVersionName()); - assertEquals(new GcpSecret("my_secret/versions/latest"), secret); - - Secret secretFoo = Secret.parseSecretOption("type:GcpSecret;version_name:foo"); - assertEquals(new GcpSecret("foo"), secretFoo); - - Secret secretMixedCase = - Secret.parseSecretOption("type:gcpsecreT;version_name:my_secret/versions/latest"); - assertEquals(new GcpSecret("my_secret/versions/latest"), secretMixedCase); - } - - @Test - public void testParseSecretOptionWithValidGcpHsmGeneratedSecret() { - String secretOption = - "type:gcphsmgeneratedsecret;project_id:my-project;location_id:global;key_ring_id:my-key-ring;key_id:my-key;job_name:my-job"; - Secret secret = Secret.parseSecretOption(secretOption); - assertTrue(secret instanceof GcpHsmGeneratedSecret); - GcpHsmGeneratedSecret hsmSecret = (GcpHsmGeneratedSecret) secret; - assertEquals("my-project", hsmSecret.getProjectId()); - assertEquals("global", hsmSecret.getLocationId()); - assertEquals("my-key-ring", hsmSecret.getKeyRingId()); - assertEquals("my-key", hsmSecret.getKeyId()); - assertEquals("HsmGeneratedSecret_my-job", hsmSecret.getSecretId()); - assertEquals( - new GcpHsmGeneratedSecret("my-project", "global", "my-key-ring", "my-key", "my-job"), - secret); - } - @Test public void testParseSecretOptionWithMissingType() { String secretOption = "version_name:my_secret/versions/latest"; @@ -86,24 +52,6 @@ public void testParseSecretOptionWithUnsupportedType() { Exception exception = assertThrows(IllegalArgumentException.class, () -> Secret.parseSecretOption(secretOption)); assertTrue(exception.getMessage().contains("Invalid secret type unsupported")); - assertTrue(exception.getMessage().contains("GcpSecret")); - assertTrue(exception.getMessage().contains("GoogleCloudSecretManager")); - } - - @Test - public void testParseSecretOptionWithInvalidGcpSecretParameter() { - String secretOption = "type:gcpsecret;invalid_param:some_value"; - Exception exception = - assertThrows(IllegalArgumentException.class, () -> Secret.parseSecretOption(secretOption)); - assertTrue(exception.getMessage().contains("Invalid secret parameter invalid_param")); - } - - @Test - public void testParseSecretOptionWithMissingSecretName() { - String secretOption = "type:gcpsecreT"; - Exception exception = - assertThrows(IllegalArgumentException.class, () -> Secret.parseSecretOption(secretOption)); - assertTrue(exception.getMessage().contains("Secret name must be specified in secret spec.")); } @Test @@ -128,21 +76,6 @@ public void testRawSecretBytes() { @Test public void testSecretFactory() { - String spec = "{\"name\": \"test-secret\", \"project\": \"proj\"}"; - - Secret secretGcp = Secret.fromJson(spec, "GoogleCloudSecretManager"); - assertTrue(secretGcp instanceof GcpSecret); - assertEquals( - "projects/proj/secrets/test-secret/versions/latest", - ((GcpSecret) secretGcp).getVersionName()); - - String singleQuotedSpec = "{'name': 'test-secret', 'project': 'proj'}"; - Secret secretSingleQuoted = Secret.fromJson(singleQuotedSpec, "GoogleCloudSecretManager"); - assertTrue(secretSingleQuoted instanceof GcpSecret); - assertEquals( - "projects/proj/secrets/test-secret/versions/latest", - ((GcpSecret) secretSingleQuoted).getVersionName()); - Secret secretRaw = Secret.fromJson("STATIC_SECRET_", null); assertTrue(secretRaw instanceof RawSecret); assertEquals("STATIC_SECRET_", secretRaw.getString()); @@ -152,23 +85,6 @@ public void testSecretFactory() { IllegalArgumentException.class, () -> Secret.fromJson("spec", "unsupported_provider")); assertTrue( exception.getMessage().contains("Unsupported secret manager: 'unsupported_provider'")); - assertTrue(exception.getMessage().contains("GoogleCloudSecretManager")); - assertTrue(exception.getMessage().contains("GcpSecret")); - - // Case-insensitive secret manager lookup in fromJson - Secret secretGcpLower = Secret.fromJson(spec, "googlecloudsecretmanager"); - assertTrue(secretGcpLower instanceof GcpSecret); - Secret secretShortLower = Secret.fromJson(spec, "gcpsecret"); - assertTrue(secretShortLower instanceof GcpSecret); - } - - @Test - public void testSecretFactoryHsm() { - String hsmSpec = - "{\"project_id\": \"p\", \"location_id\": \"l\", \"key_ring_id\": \"r\", \"key_id\": \"k\", \"job_name\": \"j\"}"; - Secret secretHsm = Secret.fromJson(hsmSpec, "GoogleCloudHsmGeneratedSecretManager"); - assertTrue(secretHsm instanceof GcpHsmGeneratedSecret); - assertEquals("p", ((GcpHsmGeneratedSecret) secretHsm).getProjectId()); } @Test @@ -188,60 +104,6 @@ public void testEquality() { assertNotEquals(raw1, raw3); assertFalse(raw1.equals("secret_value")); assertFalse(raw1.equals(null)); - - Map spec1 = new HashMap<>(); - spec1.put("name", "sec"); - spec1.put("project", "proj"); - Map spec2 = new HashMap<>(); - spec2.put("name", "sec"); - spec2.put("project", "proj"); - Map spec3 = new HashMap<>(); - spec3.put("name", "other"); - spec3.put("project", "proj"); - - GcpSecret gcp1 = GcpSecret.fromMap(spec1); - GcpSecret gcp2 = GcpSecret.fromMap(spec2); - GcpSecret gcp3 = GcpSecret.fromMap(spec3); - assertEquals(gcp1, gcp2); - assertEquals(gcp1.hashCode(), gcp2.hashCode()); - assertNotEquals(gcp1, gcp3); - assertFalse(gcp1.equals(raw1)); - assertFalse(gcp1.equals(null)); - - GcpHsmGeneratedSecret hsm1 = new GcpHsmGeneratedSecret("p", "l", "r", "k", "j"); - GcpHsmGeneratedSecret hsm2 = new GcpHsmGeneratedSecret("p", "l", "r", "k", "j"); - GcpHsmGeneratedSecret hsm3 = new GcpHsmGeneratedSecret("p", "l", "r", "k", "other"); - assertEquals(hsm1, hsm2); - assertEquals(hsm1.hashCode(), hsm2.hashCode()); - assertNotEquals(hsm1, hsm3); - assertFalse(hsm1.equals(gcp1)); - assertFalse(hsm1.equals(null)); - } - - @Test - public void testGcpSecretFromMapMissingSecretNameThrows() { - Map spec = Collections.singletonMap("project", "my-project"); - Exception exception = - assertThrows(IllegalArgumentException.class, () -> GcpSecret.fromMap(spec)); - assertTrue(exception.getMessage().contains("Secret name must be specified")); - } - - @Test - public void testGcpHsmGeneratedSecretFromMapMissingParamsThrows() { - Map spec = new HashMap<>(); - spec.put("project_id", "test-proj"); - spec.put("location_id", "global"); - Exception exception = - assertThrows(NullPointerException.class, () -> GcpHsmGeneratedSecret.fromMap(spec)); - assertTrue( - exception - .getMessage() - .contains("key_ring_id must contain a valid value for keyRingId parameter")); - } - - @Test - public void testResolveGcpProjectIdExplicit() { - assertEquals("my-proj", GcpSecret.resolveGcpProjectId("my-proj", "context")); } @Test @@ -249,14 +111,6 @@ public void testSerialization() { RawSecret raw = new RawSecret("test_secret"); RawSecret deserializedRaw = SerializableUtils.clone(raw); assertEquals(raw, deserializedRaw); - - GcpSecret gcp = new GcpSecret("projects/p/secrets/s/versions/1"); - GcpSecret deserializedGcp = SerializableUtils.clone(gcp); - assertEquals(gcp, deserializedGcp); - - GcpHsmGeneratedSecret hsm = new GcpHsmGeneratedSecret("p", "l", "r", "k", "j"); - GcpHsmGeneratedSecret deserializedHsm = SerializableUtils.clone(hsm); - assertEquals(hsm, deserializedHsm); } @Test @@ -319,19 +173,4 @@ public void testLoadSecretFactoriesDuplicateKeysFirstWins() { assertEquals(factory1, factories.get("duplicate_key")); assertEquals(Collections.singleton("duplicate_key"), supportedTypes); } - - @Test - public void testLoadServicesOrderedDiscoversSecretRegistrars() { - Iterable registrars = - ReflectHelpers.loadServicesOrdered(SecretRegistrar.class); - org.junit.Assert.assertNotNull(registrars); - boolean foundGcp = false; - for (SecretRegistrar registrar : registrars) { - if (registrar instanceof GcpSecretRegistrar) { - foundGcp = true; - break; - } - } - assertTrue("Expected GcpSecretRegistrar to be discovered", foundGcp); - } } diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/TestSecretRegistrar.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/TestSecretRegistrar.java new file mode 100644 index 000000000000..3a3b6ae83e08 --- /dev/null +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/TestSecretRegistrar.java @@ -0,0 +1,34 @@ +/* + * 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.beam.sdk.util; + +import com.google.auto.service.AutoService; +import java.nio.charset.StandardCharsets; +import java.util.Map; +import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; + +/** Test {@link SecretRegistrar} for unit and cross-language tests. */ +@AutoService(SecretRegistrar.class) +public class TestSecretRegistrar implements SecretRegistrar { + + @Override + public Map getSecretFactories() { + return ImmutableMap.of( + "TestSecret", spec -> new RawSecret(spec.get("secret").getBytes(StandardCharsets.UTF_8))); + } +} diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java index 6db2105caf79..1b4fbc10df64 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java +++ b/sdks/java/core/src/test/java/org/apache/beam/sdk/util/construction/ValidateRunnerXlangTest.java @@ -20,15 +20,11 @@ import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.equalTo; -import com.google.cloud.secretmanager.v1.ProjectName; -import com.google.cloud.secretmanager.v1.SecretManagerServiceClient; -import com.google.cloud.secretmanager.v1.SecretName; -import com.google.cloud.secretmanager.v1.SecretPayload; -import com.google.protobuf.ByteString; import java.io.IOException; import java.io.Serializable; -import java.security.SecureRandom; +import java.nio.charset.StandardCharsets; import java.util.Arrays; +import java.util.Base64; import org.apache.beam.model.pipeline.v1.ExternalTransforms; import org.apache.beam.sdk.Pipeline; import org.apache.beam.sdk.PipelineResult; @@ -55,7 +51,6 @@ import org.apache.beam.sdk.values.TypeDescriptors; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables; import org.junit.After; -import org.junit.AfterClass; import org.junit.BeforeClass; import org.junit.Rule; import org.junit.Test; @@ -89,7 +84,7 @@ * details. */ public class ValidateRunnerXlangTest { - static class ValidateRunnerXlangTestBase extends BaseExternalTest implements Serializable { + public static class ValidateRunnerXlangTestBase extends BaseExternalTest implements Serializable { // URNs for core cross-language transforms. // See https://docs.google.com/document/d/1xQp0ElIV84b8OCVz8CD2hvbiWdR8w4BvWxPTZJZA6NA for // further @@ -316,61 +311,13 @@ public void test() { @RunWith(JUnit4.class) public static class GroupByKeyWithGbekTest extends ValidateRunnerXlangTestBase { @Rule public ExpectedException thrown = ExpectedException.none(); - private static final String PROJECT_ID = "apache-beam-testing"; - private static final String SECRET_ID = "gbek-test"; - private static String gcpSecretVersionName; - private static String secretId; @BeforeClass public static void setUpClass() { - secretId = String.format("%s-%d", SECRET_ID, new SecureRandom().nextInt(10000)); - try (SecretManagerServiceClient client = SecretManagerServiceClient.create()) { - ProjectName projectName = ProjectName.of(PROJECT_ID); - SecretName secretName = SecretName.of(PROJECT_ID, secretId); - - try { - client.getSecret(secretName); - } catch (Exception e) { - com.google.cloud.secretmanager.v1.Secret secret = - com.google.cloud.secretmanager.v1.Secret.newBuilder() - .setReplication( - com.google.cloud.secretmanager.v1.Replication.newBuilder() - .setAutomatic( - com.google.cloud.secretmanager.v1.Replication.Automatic.newBuilder() - .build()) - .build()) - .build(); - client.createSecret(projectName, secretId, secret); - byte[] secretBytes = new byte[32]; - new SecureRandom().nextBytes(secretBytes); - client.addSecretVersion( - secretName, - SecretPayload.newBuilder() - .setData( - ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes))) - .build()); - } - gcpSecretVersionName = secretName.toString() + "/versions/latest"; - } catch (IOException e) { - gcpSecretVersionName = null; - return; - } expansionAddr = String.format("localhost:%s", Integer.valueOf(System.getProperty("expansionPort"))); } - @AfterClass - public static void tearDownClass() { - if (gcpSecretVersionName != null) { - try (SecretManagerServiceClient client = SecretManagerServiceClient.create()) { - SecretName secretName = SecretName.of(PROJECT_ID, secretId); - client.deleteSecret(secretName); - } catch (IOException e) { - // Do nothing. - } - } - } - @After @Override public void tearDown() { @@ -385,12 +332,10 @@ public void tearDown() { UsesPythonExpansionService.class }) public void test() { - if (gcpSecretVersionName == null) { - // Skip test if we couldn't set up secret manager - return; - } + byte[] secretBytes = "test-encryption-key-secret-12345".getBytes(StandardCharsets.UTF_8); + String base64Secret = Base64.getUrlEncoder().encodeToString(secretBytes); PipelineOptions options = TestPipeline.testingPipelineOptions(); - options.setGbek(String.format("type:gcpsecret;version_name:%s", gcpSecretVersionName)); + options.setGbek(String.format("type:testsecret;secret:%s", base64Secret)); Pipeline pipeline = Pipeline.create(options); groupByKeyTest(pipeline); PipelineResult pipelineResult = pipeline.run(); diff --git a/sdks/java/extensions/google-cloud-platform-core/build.gradle b/sdks/java/extensions/google-cloud-platform-core/build.gradle index 78cfe4739ec5..3c6612070adb 100644 --- a/sdks/java/extensions/google-cloud-platform-core/build.gradle +++ b/sdks/java/extensions/google-cloud-platform-core/build.gradle @@ -59,6 +59,12 @@ dependencies { implementation library.java.jackson_core implementation library.java.jackson_databind permitUnusedDeclared library.java.jackson_databind // BEAM-11761 + implementation library.java.google_cloud_kms + implementation library.java.proto_google_cloud_kms_v1 + implementation library.java.google_cloud_tink + implementation library.java.google_cloud_secret_manager + implementation library.java.proto_google_cloud_secret_manager_v1 + implementation library.java.protobuf_java testImplementation project(path: ":sdks:java:core", configuration: "shadowTest") testImplementation library.java.mockito_core testRuntimeOnly library.java.slf4j_jdk14 diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecret.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpHsmGeneratedSecret.java similarity index 96% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecret.java rename to sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpHsmGeneratedSecret.java index 9a62bc49d3ca..c1227cbff560 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecret.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpHsmGeneratedSecret.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util; +package org.apache.beam.sdk.extensions.gcp.util; import com.google.api.gax.rpc.AlreadyExistsException; import com.google.api.gax.rpc.NotFoundException; @@ -43,15 +43,15 @@ import java.util.Map; import java.util.Objects; import java.util.Set; +import org.apache.beam.sdk.util.Secret; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions; import org.checkerframework.checker.nullness.qual.Nullable; import org.slf4j.Logger; import org.slf4j.LoggerFactory; /** - * A {@link org.apache.beam.sdk.util.Secret} manager implementation that generates a secret using - * entropy from a GCP HSM key and stores it in Google Cloud Secret Manager. If the secret already - * exists, it will be retrieved. + * A {@link Secret} manager implementation that generates a secret using entropy from a GCP HSM key + * and stores it in Google Cloud Secret Manager. If the secret already exists, it will be retrieved. */ public class GcpHsmGeneratedSecret extends Secret { private static final Logger LOG = LoggerFactory.getLogger(GcpHsmGeneratedSecret.class); diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecretRegistrar.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpHsmGeneratedSecretRegistrar.java similarity index 93% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecretRegistrar.java rename to sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpHsmGeneratedSecretRegistrar.java index 232fe7dfa835..ecdaf521ff07 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpHsmGeneratedSecretRegistrar.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpHsmGeneratedSecretRegistrar.java @@ -15,10 +15,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util; +package org.apache.beam.sdk.extensions.gcp.util; import com.google.auto.service.AutoService; import java.util.Map; +import org.apache.beam.sdk.util.SecretRegistrar; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; /** {@link AutoService} registrar for the {@link GcpHsmGeneratedSecret}. */ diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecret.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecret.java similarity index 98% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecret.java rename to sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecret.java index 9aaa61bc3e00..083298305997 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecret.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecret.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util; +package org.apache.beam.sdk.extensions.gcp.util; import com.google.cloud.secretmanager.v1.AccessSecretVersionResponse; import com.google.cloud.secretmanager.v1.SecretManagerServiceClient; @@ -29,6 +29,7 @@ import java.util.Map; import java.util.Objects; import java.util.Set; +import org.apache.beam.sdk.util.Secret; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Strings; import org.checkerframework.checker.nullness.qual.Nullable; diff --git a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecretRegistrar.java b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretRegistrar.java similarity index 93% rename from sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecretRegistrar.java rename to sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretRegistrar.java index 61b31332e6dd..600ffa53391f 100644 --- a/sdks/java/core/src/main/java/org/apache/beam/sdk/util/GcpSecretRegistrar.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/main/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretRegistrar.java @@ -15,10 +15,11 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util; +package org.apache.beam.sdk.extensions.gcp.util; import com.google.auto.service.AutoService; import java.util.Map; +import org.apache.beam.sdk.util.SecretRegistrar; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap; /** {@link AutoService} registrar for the {@link GcpSecret}. */ diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/transforms/GcpGroupByEncryptedKeyTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/transforms/GcpGroupByEncryptedKeyTest.java new file mode 100644 index 000000000000..9232677f56c0 --- /dev/null +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/transforms/GcpGroupByEncryptedKeyTest.java @@ -0,0 +1,299 @@ +/* + * 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.beam.sdk.extensions.gcp.transforms; + +import static org.junit.Assert.assertThrows; + +import com.google.cloud.secretmanager.v1.ProjectName; +import com.google.cloud.secretmanager.v1.SecretManagerServiceClient; +import com.google.cloud.secretmanager.v1.SecretName; +import com.google.cloud.secretmanager.v1.SecretPayload; +import com.google.protobuf.ByteString; +import java.io.IOException; +import java.io.Serializable; +import java.security.SecureRandom; +import java.util.Arrays; +import java.util.List; +import java.util.stream.Collectors; +import java.util.stream.StreamSupport; +import org.apache.beam.sdk.coders.BigEndianIntegerCoder; +import org.apache.beam.sdk.coders.KvCoder; +import org.apache.beam.sdk.coders.NullableCoder; +import org.apache.beam.sdk.coders.StringUtf8Coder; +import org.apache.beam.sdk.coders.VarIntCoder; +import org.apache.beam.sdk.extensions.gcp.util.GcpHsmGeneratedSecret; +import org.apache.beam.sdk.extensions.gcp.util.GcpSecret; +import org.apache.beam.sdk.testing.NeedsRunner; +import org.apache.beam.sdk.testing.PAssert; +import org.apache.beam.sdk.testing.TestPipeline; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.GroupByEncryptedKey; +import org.apache.beam.sdk.transforms.GroupByKey; +import org.apache.beam.sdk.transforms.MapElements; +import org.apache.beam.sdk.transforms.SimpleFunction; +import org.apache.beam.sdk.util.Secret; +import org.apache.beam.sdk.values.KV; +import org.apache.beam.sdk.values.PCollection; +import org.checkerframework.checker.nullness.qual.Nullable; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Rule; +import org.junit.Test; +import org.junit.experimental.categories.Category; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests for {@link GroupByEncryptedKey} and GBEK pipeline options using GCP Secrets. */ +@RunWith(JUnit4.class) +public class GcpGroupByEncryptedKeyTest implements Serializable { + + @Rule public transient TestPipeline p = TestPipeline.create(); + + private static final String PROJECT_ID = "apache-beam-testing"; + private static final String SECRET_ID = "gbek-test"; + private static Secret gcpSecret; + private static String gcpSecretVersionName; + private static Secret gcpHsmGeneratedSecret; + private static final String KEY_RING_ID = "gbek-test-key-ring"; + private static final String KEY_ID = "gbek-test-key"; + + @BeforeClass + public static void setup() throws IOException { + SecretManagerServiceClient client; + try { + client = SecretManagerServiceClient.create(); + } catch (IOException e) { + gcpSecretVersionName = null; + return; + } + ProjectName projectName = ProjectName.of(PROJECT_ID); + SecretName secretName = SecretName.of(PROJECT_ID, SECRET_ID); + + try { + client.getSecret(secretName); + } catch (Exception e) { + com.google.cloud.secretmanager.v1.Secret secret = + com.google.cloud.secretmanager.v1.Secret.newBuilder() + .setReplication( + com.google.cloud.secretmanager.v1.Replication.newBuilder() + .setAutomatic( + com.google.cloud.secretmanager.v1.Replication.Automatic.newBuilder() + .build()) + .build()) + .build(); + client.createSecret(projectName, SECRET_ID, secret); + byte[] secretBytes = new byte[32]; + new SecureRandom().nextBytes(secretBytes); + client.addSecretVersion( + secretName, + SecretPayload.newBuilder() + .setData(ByteString.copyFrom(java.util.Base64.getUrlEncoder().encode(secretBytes))) + .build()); + } + gcpSecretVersionName = secretName.toString() + "/versions/latest"; + gcpSecret = new GcpSecret(gcpSecretVersionName); + + try { + com.google.cloud.kms.v1.KeyManagementServiceClient kmsClient = + com.google.cloud.kms.v1.KeyManagementServiceClient.create(); + String locationId = "global"; + com.google.cloud.kms.v1.KeyRingName keyRingName = + com.google.cloud.kms.v1.KeyRingName.of(PROJECT_ID, locationId, KEY_RING_ID); + com.google.cloud.kms.v1.LocationName locationName = + com.google.cloud.kms.v1.LocationName.of(PROJECT_ID, locationId); + try { + kmsClient.getKeyRing(keyRingName); + } catch (Exception e) { + kmsClient.createKeyRing( + locationName, KEY_RING_ID, com.google.cloud.kms.v1.KeyRing.newBuilder().build()); + } + + com.google.cloud.kms.v1.CryptoKeyName keyName = + com.google.cloud.kms.v1.CryptoKeyName.of(PROJECT_ID, locationId, KEY_RING_ID, KEY_ID); + try { + kmsClient.getCryptoKey(keyName); + } catch (Exception e) { + com.google.cloud.kms.v1.CryptoKey key = + com.google.cloud.kms.v1.CryptoKey.newBuilder() + .setPurpose(com.google.cloud.kms.v1.CryptoKey.CryptoKeyPurpose.ENCRYPT_DECRYPT) + .build(); + kmsClient.createCryptoKey(keyRingName, KEY_ID, key); + } + gcpHsmGeneratedSecret = + new GcpHsmGeneratedSecret( + PROJECT_ID, + locationId, + KEY_RING_ID, + KEY_ID, + String.format("gbek-test-job-%d", new SecureRandom().nextInt(10000))); + // Validate we have crypto permissions or skip these tests. + gcpHsmGeneratedSecret.getSecretBytes(); + } catch (Exception e) { + gcpHsmGeneratedSecret = null; + } + } + + @AfterClass + public static void tearDown() throws IOException { + if (gcpSecretVersionName != null) { + SecretManagerServiceClient client = SecretManagerServiceClient.create(); + SecretName secretName = SecretName.of(PROJECT_ID, SECRET_ID); + client.deleteSecret(secretName); + } + } + + @Test + @Category(NeedsRunner.class) + public void testGroupByKeyGcpSecret() { + if (gcpSecret == null) { + return; + } + List> ungroupedPairs = + Arrays.asList( + KV.of(null, 3), + KV.of("k1", 3), + KV.of("k5", Integer.MAX_VALUE), + KV.of("k5", Integer.MIN_VALUE), + KV.of("k2", 66), + KV.of("k1", 4), + KV.of(null, 5), + KV.of("k2", -33), + KV.of("k3", 0)); + + PCollection> input = + p.apply( + Create.of(ungroupedPairs) + .withCoder(KvCoder.of(NullableCoder.of(StringUtf8Coder.of()), VarIntCoder.of()))); + + PCollection>> output = + input.apply(GroupByEncryptedKey.create(gcpSecret)); + + PAssert.that(output.apply("Sort", MapElements.via(new SortValues()))) + .containsInAnyOrder( + KV.of("k1", Arrays.asList(3, 4)), + KV.of(null, Arrays.asList(3, 5)), + KV.of("k5", Arrays.asList(Integer.MIN_VALUE, Integer.MAX_VALUE)), + KV.of("k2", Arrays.asList(-33, 66)), + KV.of("k3", Arrays.asList(0))); + + p.run(); + } + + @Test + @Category(NeedsRunner.class) + public void testGroupByKeyGcpSecretThrows() { + Secret badGcpSecret = new GcpSecret("bad_path/versions/latest"); + p.apply(Create.of(KV.of("k1", 1))) + .apply(GroupByEncryptedKey.create(badGcpSecret)); + assertThrows(RuntimeException.class, () -> p.run()); + } + + @Test + @Category(NeedsRunner.class) + public void testGroupByKeyGcpHsmGeneratedSecret() { + if (gcpHsmGeneratedSecret == null) { + return; + } + List> ungroupedPairs = + Arrays.asList( + KV.of(null, 3), + KV.of("k1", 3), + KV.of("k5", Integer.MAX_VALUE), + KV.of("k5", Integer.MIN_VALUE), + KV.of("k2", 66), + KV.of("k1", 4), + KV.of(null, 5), + KV.of("k2", -33), + KV.of("k3", 0)); + + PCollection> input = + p.apply( + Create.of(ungroupedPairs) + .withCoder(KvCoder.of(NullableCoder.of(StringUtf8Coder.of()), VarIntCoder.of()))); + + PCollection>> output = + input.apply(GroupByEncryptedKey.create(gcpHsmGeneratedSecret)); + + PAssert.that(output.apply("Sort", MapElements.via(new SortValues()))) + .containsInAnyOrder( + KV.of("k1", Arrays.asList(3, 4)), + KV.of(null, Arrays.asList(3, 5)), + KV.of("k5", Arrays.asList(Integer.MIN_VALUE, Integer.MAX_VALUE)), + KV.of("k2", Arrays.asList(-33, 66)), + KV.of("k3", Arrays.asList(0))); + + p.run(); + } + + @Test + @Category(NeedsRunner.class) + public void testGroupByKeyWithValidGcpSecretOption() { + if (gcpSecretVersionName == null) { + return; + } + List> ungroupedPairs = + Arrays.asList( + KV.of("k1", 3), + KV.of("k5", Integer.MAX_VALUE), + KV.of("k5", Integer.MIN_VALUE), + KV.of("k2", 66), + KV.of("k1", 4), + KV.of("k2", -33), + KV.of("k3", 0)); + + PCollection> input = + p.apply( + Create.of(ungroupedPairs) + .withCoder(KvCoder.of(StringUtf8Coder.of(), BigEndianIntegerCoder.of()))); + + p.getOptions().setGbek(String.format("type:gcpsecret;version_name:%s", gcpSecretVersionName)); + PCollection>> output = input.apply(GroupByKey.create()); + + PAssert.that(output.apply("Sort", MapElements.via(new SortValues()))) + .containsInAnyOrder( + KV.of("k1", Arrays.asList(3, 4)), + KV.of("k5", Arrays.asList(Integer.MIN_VALUE, Integer.MAX_VALUE)), + KV.of("k2", Arrays.asList(66, -33)), + KV.of("k3", Arrays.asList(0))); + + p.run(); + } + + @Test + @Category(NeedsRunner.class) + public void testGroupByKeyWithInvalidGcpSecretOption() { + if (gcpSecretVersionName == null) { + return; + } + p.getOptions().setGbek("type:gcpsecret;version_name:bad_path/versions/latest"); + p.apply(Create.of(KV.of("k1", 1))).apply(GroupByKey.create()); + assertThrows(RuntimeException.class, () -> p.run()); + } + + private static class SortValues + extends SimpleFunction>, KV>> { + @Override + public KV> apply(KV> input) { + List sorted = + StreamSupport.stream(input.getValue().spliterator(), false) + .sorted() + .collect(Collectors.toList()); + return KV.of(input.getKey(), sorted); + } + } +} diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/transforms/GcpGroupByKeyIT.java similarity index 97% rename from sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java rename to sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/transforms/GcpGroupByKeyIT.java index 431bdf448bea..e8d3466cf097 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/transforms/GroupByKeyIT.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/transforms/GcpGroupByKeyIT.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.transforms; +package org.apache.beam.sdk.extensions.gcp.transforms; import com.google.cloud.kms.v1.CryptoKey; import com.google.cloud.kms.v1.CryptoKeyName; @@ -34,10 +34,15 @@ import org.apache.beam.sdk.coders.KvCoder; import org.apache.beam.sdk.coders.StringUtf8Coder; import org.apache.beam.sdk.coders.VarIntCoder; +import org.apache.beam.sdk.extensions.gcp.util.GcpHsmGeneratedSecret; import org.apache.beam.sdk.options.PipelineOptions; import org.apache.beam.sdk.testing.PAssert; import org.apache.beam.sdk.testing.TestPipeline; -import org.apache.beam.sdk.util.GcpHsmGeneratedSecret; +import org.apache.beam.sdk.transforms.Combine; +import org.apache.beam.sdk.transforms.Create; +import org.apache.beam.sdk.transforms.GroupByKey; +import org.apache.beam.sdk.transforms.Redistribute; +import org.apache.beam.sdk.transforms.Sum; import org.apache.beam.sdk.values.KV; import org.apache.beam.sdk.values.PCollection; import org.junit.AfterClass; @@ -50,7 +55,7 @@ /** Integration test for GroupByKey transforms and some other transforms which use GBK. */ @RunWith(JUnit4.class) -public class GroupByKeyIT { +public class GcpGroupByKeyIT { @Rule public ExpectedException thrown = ExpectedException.none(); private static final String PROJECT_ID = "apache-beam-testing"; diff --git a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/GcpSecretRegistrarTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretRegistrarTest.java similarity index 96% rename from sdks/java/core/src/test/java/org/apache/beam/sdk/util/GcpSecretRegistrarTest.java rename to sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretRegistrarTest.java index e95483dd9bf7..f8d3708f8d32 100644 --- a/sdks/java/core/src/test/java/org/apache/beam/sdk/util/GcpSecretRegistrarTest.java +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretRegistrarTest.java @@ -15,7 +15,7 @@ * See the License for the specific language governing permissions and * limitations under the License. */ -package org.apache.beam.sdk.util; +package org.apache.beam.sdk.extensions.gcp.util; import static org.hamcrest.MatcherAssert.assertThat; import static org.hamcrest.Matchers.hasItems; @@ -23,6 +23,7 @@ import java.util.Map; import java.util.ServiceLoader; +import org.apache.beam.sdk.util.SecretRegistrar; import org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Lists; import org.junit.Test; import org.junit.runner.RunWith; diff --git a/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretTest.java b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretTest.java new file mode 100644 index 000000000000..c493c642dc25 --- /dev/null +++ b/sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretTest.java @@ -0,0 +1,225 @@ +/* + * 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.beam.sdk.extensions.gcp.util; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertNotEquals; +import static org.junit.Assert.assertThrows; +import static org.junit.Assert.assertTrue; + +import java.util.Collections; +import java.util.HashMap; +import java.util.Map; +import org.apache.beam.sdk.util.RawSecret; +import org.apache.beam.sdk.util.Secret; +import org.apache.beam.sdk.util.SerializableUtils; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +/** Tests for {@link GcpSecret} and {@link GcpHsmGeneratedSecret}. */ +@RunWith(JUnit4.class) +public class GcpSecretTest { + + @Test + public void testParseSecretOptionWithValidGcpSecret() { + String secretOption = "type:gcpsecret;version_name:my_secret/versions/latest"; + Secret secret = Secret.parseSecretOption(secretOption); + assertTrue(secret instanceof GcpSecret); + assertEquals("my_secret/versions/latest", ((GcpSecret) secret).getVersionName()); + assertEquals(new GcpSecret("my_secret/versions/latest"), secret); + + Secret secretFoo = Secret.parseSecretOption("type:GcpSecret;version_name:foo"); + assertEquals(new GcpSecret("foo"), secretFoo); + + Secret secretMixedCase = + Secret.parseSecretOption("type:gcpsecreT;version_name:my_secret/versions/latest"); + assertEquals(new GcpSecret("my_secret/versions/latest"), secretMixedCase); + } + + @Test + public void testParseSecretOptionWithValidGcpHsmGeneratedSecret() { + String secretOption = + "type:gcphsmgeneratedsecret;project_id:my-project;location_id:global;key_ring_id:my-key-ring;key_id:my-key;job_name:my-job"; + Secret secret = Secret.parseSecretOption(secretOption); + assertTrue(secret instanceof GcpHsmGeneratedSecret); + GcpHsmGeneratedSecret hsmSecret = (GcpHsmGeneratedSecret) secret; + assertEquals("my-project", hsmSecret.getProjectId()); + assertEquals("global", hsmSecret.getLocationId()); + assertEquals("my-key-ring", hsmSecret.getKeyRingId()); + assertEquals("my-key", hsmSecret.getKeyId()); + assertEquals("HsmGeneratedSecret_my-job", hsmSecret.getSecretId()); + assertEquals( + new GcpHsmGeneratedSecret("my-project", "global", "my-key-ring", "my-key", "my-job"), + secret); + } + + @Test + public void testParseSecretOptionWithInvalidGcpSecretParameter() { + String secretOption = "type:gcpsecret;invalid_param:some_value"; + Exception exception = + assertThrows(IllegalArgumentException.class, () -> Secret.parseSecretOption(secretOption)); + assertTrue(exception.getMessage().contains("Invalid secret parameter invalid_param")); + } + + @Test + public void testParseSecretOptionWithMissingSecretName() { + String secretOption = "type:gcpsecreT"; + Exception exception = + assertThrows(IllegalArgumentException.class, () -> Secret.parseSecretOption(secretOption)); + assertTrue(exception.getMessage().contains("Secret name must be specified in secret spec.")); + } + + @Test + public void testSecretFactoryGcp() { + String spec = "{\"name\": \"test-secret\", \"project\": \"proj\"}"; + + Secret secretGcp = Secret.fromJson(spec, "GoogleCloudSecretManager"); + assertTrue(secretGcp instanceof GcpSecret); + assertEquals( + "projects/proj/secrets/test-secret/versions/latest", + ((GcpSecret) secretGcp).getVersionName()); + + String singleQuotedSpec = "{'name': 'test-secret', 'project': 'proj'}"; + Secret secretSingleQuoted = Secret.fromJson(singleQuotedSpec, "GoogleCloudSecretManager"); + assertTrue(secretSingleQuoted instanceof GcpSecret); + assertEquals( + "projects/proj/secrets/test-secret/versions/latest", + ((GcpSecret) secretSingleQuoted).getVersionName()); + + // Case-insensitive secret manager lookup in fromJson + Secret secretGcpLower = Secret.fromJson(spec, "googlecloudsecretmanager"); + assertTrue(secretGcpLower instanceof GcpSecret); + Secret secretShortLower = Secret.fromJson(spec, "gcpsecret"); + assertTrue(secretShortLower instanceof GcpSecret); + + Exception exception = + assertThrows( + IllegalArgumentException.class, () -> Secret.fromJson("spec", "unsupported_provider")); + assertTrue( + exception.getMessage().contains("Unsupported secret manager: 'unsupported_provider'")); + assertTrue(exception.getMessage().contains("GoogleCloudSecretManager")); + assertTrue(exception.getMessage().contains("GcpSecret")); + } + + @Test + public void testParseSecretOptionWithUnsupportedTypeListsGcpSecrets() { + String secretOption = "type:unsupported;version_name:my_secret/versions/latest"; + Exception exception = + assertThrows(IllegalArgumentException.class, () -> Secret.parseSecretOption(secretOption)); + assertTrue(exception.getMessage().contains("Invalid secret type unsupported")); + assertTrue(exception.getMessage().contains("GcpSecret")); + assertTrue(exception.getMessage().contains("GoogleCloudSecretManager")); + } + + @Test + public void testLoadServicesOrderedDiscoversSecretRegistrars() { + Iterable registrars = + org.apache.beam.sdk.util.common.ReflectHelpers.loadServicesOrdered( + org.apache.beam.sdk.util.SecretRegistrar.class); + org.junit.Assert.assertNotNull(registrars); + boolean foundGcp = false; + for (org.apache.beam.sdk.util.SecretRegistrar registrar : registrars) { + if (registrar instanceof GcpSecretRegistrar) { + foundGcp = true; + break; + } + } + assertTrue("Expected GcpSecretRegistrar to be discovered", foundGcp); + } + + @Test + public void testSecretFactoryHsm() { + String hsmSpec = + "{\"project_id\": \"p\", \"location_id\": \"l\", \"key_ring_id\": \"r\", \"key_id\": \"k\", \"job_name\": \"j\"}"; + Secret secretHsm = Secret.fromJson(hsmSpec, "GoogleCloudHsmGeneratedSecretManager"); + assertTrue(secretHsm instanceof GcpHsmGeneratedSecret); + assertEquals("p", ((GcpHsmGeneratedSecret) secretHsm).getProjectId()); + } + + @Test + public void testEquality() { + RawSecret raw1 = new RawSecret("secret_value"); + + Map spec1 = new HashMap<>(); + spec1.put("name", "sec"); + spec1.put("project", "proj"); + Map spec2 = new HashMap<>(); + spec2.put("name", "sec"); + spec2.put("project", "proj"); + Map spec3 = new HashMap<>(); + spec3.put("name", "other"); + spec3.put("project", "proj"); + + GcpSecret gcp1 = GcpSecret.fromMap(spec1); + GcpSecret gcp2 = GcpSecret.fromMap(spec2); + GcpSecret gcp3 = GcpSecret.fromMap(spec3); + assertEquals(gcp1, gcp2); + assertEquals(gcp1.hashCode(), gcp2.hashCode()); + assertNotEquals(gcp1, gcp3); + assertFalse(gcp1.equals(raw1)); + assertFalse(gcp1.equals(null)); + + GcpHsmGeneratedSecret hsm1 = new GcpHsmGeneratedSecret("p", "l", "r", "k", "j"); + GcpHsmGeneratedSecret hsm2 = new GcpHsmGeneratedSecret("p", "l", "r", "k", "j"); + GcpHsmGeneratedSecret hsm3 = new GcpHsmGeneratedSecret("p", "l", "r", "k", "other"); + assertEquals(hsm1, hsm2); + assertEquals(hsm1.hashCode(), hsm2.hashCode()); + assertNotEquals(hsm1, hsm3); + assertFalse(hsm1.equals(gcp1)); + assertFalse(hsm1.equals(null)); + } + + @Test + public void testGcpSecretFromMapMissingSecretNameThrows() { + Map spec = Collections.singletonMap("project", "my-project"); + Exception exception = + assertThrows(IllegalArgumentException.class, () -> GcpSecret.fromMap(spec)); + assertTrue(exception.getMessage().contains("Secret name must be specified")); + } + + @Test + public void testGcpHsmGeneratedSecretFromMapMissingParamsThrows() { + Map spec = new HashMap<>(); + spec.put("project_id", "test-proj"); + spec.put("location_id", "global"); + Exception exception = + assertThrows(NullPointerException.class, () -> GcpHsmGeneratedSecret.fromMap(spec)); + assertTrue( + exception + .getMessage() + .contains("key_ring_id must contain a valid value for keyRingId parameter")); + } + + @Test + public void testResolveGcpProjectIdExplicit() { + assertEquals("my-proj", GcpSecret.resolveGcpProjectId("my-proj", "context")); + } + + @Test + public void testSerialization() { + GcpSecret gcp = new GcpSecret("projects/p/secrets/s/versions/1"); + GcpSecret deserializedGcp = SerializableUtils.clone(gcp); + assertEquals(gcp, deserializedGcp); + + GcpHsmGeneratedSecret hsm = new GcpHsmGeneratedSecret("p", "l", "r", "k", "j"); + GcpHsmGeneratedSecret deserializedHsm = SerializableUtils.clone(hsm); + assertEquals(hsm, deserializedHsm); + } +} diff --git a/sdks/java/io/components/build.gradle b/sdks/java/io/components/build.gradle index 973420098348..4d3fe55033da 100644 --- a/sdks/java/io/components/build.gradle +++ b/sdks/java/io/components/build.gradle @@ -37,7 +37,10 @@ dependencies { implementation library.java.vendored_guava_32_1_2_jre implementation library.java.vendored_grpc_1_69_0 + testImplementation platform(library.java.google_cloud_platform_libraries_bom) testImplementation project(path: ":sdks:java:core", configuration: "shadowTest") + testImplementation library.java.grpc_inprocess + testRuntimeOnly library.java.grpc_netty_shaded testImplementation library.java.guava_testlib testImplementation library.java.junit testImplementation library.java.hamcrest diff --git a/sdks/python/apache_beam/runners/portability/expansion_service_test.py b/sdks/python/apache_beam/runners/portability/expansion_service_test.py index b0b0b2dd2bfa..67ca65018803 100644 --- a/sdks/python/apache_beam/runners/portability/expansion_service_test.py +++ b/sdks/python/apache_beam/runners/portability/expansion_service_test.py @@ -39,8 +39,14 @@ from apache_beam.transforms import ptransform from apache_beam.transforms.environments import PyPIArtifactRegistry from apache_beam.transforms.external import ImplicitSchemaPayloadBuilder +from apache_beam.utils import secret as beam_secret from apache_beam.utils import thread_pool_executor +# Register TestSecret for cross-language GBEK validation tests. +beam_secret._SECRET_TYPE_TO_SECRET_MANAGER["testsecret"] = "TestSecretManager" +beam_secret._SECRET_CLASSES["testsecretmanager"] = ( + lambda secret: beam_secret.RawSecret(secret.encode("utf-8"))) + # This script provides an expansion service and example ptransforms for running # external transform test cases. See external_test.py for details.