Allow ':' in secret option values and reject malformed parameters

sdks/java/core/src/main/java/org/apache/beam/sdk/util/Secret.java

@@ -198,6 +198,9 @@ public byte[] getBytes() {
    * <p>The secret string should be formatted like:
    * 'type:&lt;secret_type&gt;;&lt;secret_param&gt;:&lt;value&gt;'
    *
+   * <p>Only the first ':' of each parameter separates the name from the value, so values may
+   * contain ':' (e.g. ARNs or URLs), but not ';'.
+   *
    * <p>For example, 'type:GcpSecret;version_name:my_secret/versions/latest' would return a
    * GcpSecret initialized with 'my_secret/versions/latest'.
    */
@@ -207,10 +210,16 @@ public static Secret parseSecretOption(String secretOption) {
     }
     Map<String, String> paramMap = new HashMap<>();
     for (String param : secretOption.split(";", -1)) {
+      if (param.isEmpty()) {
+        continue;
+      }
       String[] parts = param.split(":", 2);
-      if (parts.length == 2) {
-        paramMap.put(parts[0], parts[1]);
+      if (parts.length != 2) {
+        throw new IllegalArgumentException(
+            String.format(
+                "Invalid secret parameter '%s', expected '<secret_param>:<value>'", param));
       }
+      paramMap.put(parts[0], parts[1]);
     }
 
     if (!paramMap.containsKey("type")) {

sdks/java/core/src/test/java/org/apache/beam/sdk/util/SecretTest.java

@@ -54,6 +54,26 @@ public void testParseSecretOptionWithUnsupportedType() {
     assertTrue(exception.getMessage().contains("Invalid secret type unsupported"));
   }
 
+  @Test
+  public void testParseSecretOptionWithMalformedParameter() {
+    String secretOption = "type:unsupported;version_name";
+    Exception exception =
+        assertThrows(IllegalArgumentException.class, () -> Secret.parseSecretOption(secretOption));
+    assertTrue(exception.getMessage().contains("Invalid secret parameter 'version_name'"));
+  }
+
+  @Test
+  public void testParseSecretOptionWithColonInValue() {
+    // Only the first ':' separates the parameter name from its value.
+    assertEquals(
+        new RawSecret("arn:aws:secretsmanager:us-east-1:123:secret:my-secret"),
+        Secret.parseSecretOption(
+            "type:TestSecret;secret:arn:aws:secretsmanager:us-east-1:123:secret:my-secret"));
+    assertEquals(
+        new RawSecret("http://localhost:4566"),
+        Secret.parseSecretOption("type:TestSecret;secret:http://localhost:4566;"));
+  }
+
   @Test
   public void testRawSecretStr() {
     Secret secret = new RawSecret("STATIC_SECRET_");

sdks/java/extensions/google-cloud-platform-core/src/test/java/org/apache/beam/sdk/extensions/gcp/util/GcpSecretTest.java

@@ -51,6 +51,10 @@ public void testParseSecretOptionWithValidGcpSecret() {
     Secret secretMixedCase =
         Secret.parseSecretOption("type:gcpsecreT;version_name:my_secret/versions/latest");
     assertEquals(new GcpSecret("my_secret/versions/latest"), secretMixedCase);
+
+    // Only the first ':' separates the parameter name from its value.
+    Secret secretWithColon = Secret.parseSecretOption("type:gcpsecret;version_name:a:b/c;");
+    assertEquals(new GcpSecret("a:b/c"), secretWithColon);
   }
 
   @Test

sdks/python/apache_beam/utils/secret.py

@@ -105,6 +105,9 @@ def parse_secret_option(
     The secret string should be formatted like:
     'type:<secret_type>;<secret_param>:<value>'
 
+    Only the first ':' of each parameter separates the name from the value, so
+    values may contain ':' (e.g. ARNs or URLs), but not ';'.
+
     For example, 'type:GcpSecret;version_name:my_secret/versions/latest'
     would return a GcpSecret initialized with 'my_secret/versions/latest'.
 
@@ -115,9 +118,14 @@ def parse_secret_option(
     """
     param_map = {}
     for param in secret.split(';'):
-      parts = param.split(':')
-      if len(parts) == 2:
-        param_map[parts[0]] = parts[1]
+      if not param:
+        continue
+      if ':' not in param:
+        raise ValueError(
+            f"Invalid secret parameter '{param}', expected "
+            "'<secret_param>:<value>'")
+      key, value = param.split(':', 1)
+      param_map[key] = value
 
     if 'type' not in param_map:
       raise ValueError('Secret string must contain a valid type parameter')

sdks/python/apache_beam/utils/secret_test.py

@@ -59,6 +59,18 @@ class SecretTest(unittest.TestCase):
       param(
           secret_string='type:awssecreT;name:my_aws_secret;version:v1',
           secret=AwsSecret('my_aws_secret', version_id='v1')),
+      param(
+          secret_string=(
+              'type:AwsSecret;'
+              'name:arn:aws:secretsmanager:us-east-1:123:secret:my-secret'),
+          secret=AwsSecret(
+              'arn:aws:secretsmanager:us-east-1:123:secret:my-secret')),
+      param(
+          secret_string=(
+              'type:AwsSecret;name:my_aws_secret;'
+              'endpoint_url:http://localhost:4566;'),
+          secret=AwsSecret(
+              'my_aws_secret', endpoint_url='http://localhost:4566')),
   ])
   def test_secret_manager_parses_correctly(self, secret_string, secret):
     self.assertEqual(secret, Secret.parse_secret_option(secret_string))
@@ -73,6 +85,9 @@ def test_secret_manager_parses_correctly(self, secret_string, secret):
       param(
           secret_string='type:gcpsecreT;version_name:foo;extra:val',
           exception_str='Invalid secret parameter extra'),
+      param(
+          secret_string='type:gcpsecreT;version_name:foo;extra',
+          exception_str="Invalid secret parameter 'extra'"),
       param(
           secret_string='type:awssecreT',
           exception_str='Secret ID or name must be specified in secret spec'),