Refactor Java Secret classes to align with Python SDK

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

@@ -34,7 +34,17 @@
 import java.io.IOException;
 import java.security.GeneralSecurityException;
 import java.security.SecureRandom;
+import java.util.ArrayList;
+import java.util.Arrays;
 import java.util.Base64;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+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;
 
@@ -43,7 +53,7 @@
  * 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 implements Secret {
+public class GcpHsmGeneratedSecret extends Secret {
   private static final Logger LOG = LoggerFactory.getLogger(GcpHsmGeneratedSecret.class);
   private final String projectId;
   private final String locationId;
@@ -62,6 +72,38 @@ public GcpHsmGeneratedSecret(
     this.secretId = "HsmGeneratedSecret_" + jobName;
   }
 
+  /** Initialize GcpHsmGeneratedSecret from a map specification. */
+  static GcpHsmGeneratedSecret fromMap(Map<String, String> specMap) {
+    Set<String> allowedKeys =
+        new HashSet<>(
+            Arrays.asList("project_id", "location_id", "key_ring_id", "key_id", "job_name"));
+    Set<String> invalid = new HashSet<>(specMap.keySet());
+    invalid.removeAll(allowedKeys);
+    if (!invalid.isEmpty()) {
+      List<String> sortedInvalid = new ArrayList<>(invalid);
+      Collections.sort(sortedInvalid);
+      throw new IllegalArgumentException(
+          "Invalid secret parameter " + String.join(", ", sortedInvalid));
+    }
+    String locationId =
+        Preconditions.checkNotNull(
+            specMap.get("location_id"),
+            "location_id must contain a valid value for locationId parameter");
+    String keyRingId =
+        Preconditions.checkNotNull(
+            specMap.get("key_ring_id"),
+            "key_ring_id must contain a valid value for keyRingId parameter");
+    String keyId =
+        Preconditions.checkNotNull(
+            specMap.get("key_id"), "key_id must contain a valid value for keyId parameter");
+    String jobName =
+        Preconditions.checkNotNull(
+            specMap.get("job_name"), "job_name must contain a valid value for jobName parameter");
+    String projectId =
+        GcpSecret.resolveGcpProjectId(specMap.get("project_id"), "job '" + jobName + "'");
+    return new GcpHsmGeneratedSecret(projectId, locationId, keyRingId, keyId, jobName);
+  }
+
   /**
    * Returns the secret as a byte array. Assumes that the current active service account has
    * permissions to read the secret.
@@ -188,4 +230,25 @@ public String getKeyId() {
   public String getSecretId() {
     return secretId;
   }
+
+  @Override
+  public boolean equals(@Nullable Object obj) {
+    if (this == obj) {
+      return true;
+    }
+    if (!(obj instanceof GcpHsmGeneratedSecret)) {
+      return false;
+    }
+    GcpHsmGeneratedSecret other = (GcpHsmGeneratedSecret) obj;
+    return Objects.equals(this.projectId, other.projectId)
+        && Objects.equals(this.locationId, other.locationId)
+        && Objects.equals(this.keyRingId, other.keyRingId)
+        && Objects.equals(this.keyId, other.keyId)
+        && Objects.equals(this.secretId, other.secretId);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(projectId, locationId, keyRingId, keyId, secretId);
+  }
 }

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

@@ -21,11 +21,25 @@
 import com.google.cloud.secretmanager.v1.SecretManagerServiceClient;
 import com.google.cloud.secretmanager.v1.SecretVersionName;
 import java.io.IOException;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+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;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 /**
  * A {@link Secret} manager implementation that retrieves secrets from Google Cloud Secret Manager.
  */
-public class GcpSecret implements Secret {
+public class GcpSecret extends Secret {
+  private static final Logger LOG = LoggerFactory.getLogger(GcpSecret.class);
   private final String versionName;
 
   /**
@@ -39,6 +53,76 @@ public GcpSecret(String versionName) {
     this.versionName = versionName;
   }
 
+  /** Initialize GcpSecret from a map specification. */
+  static GcpSecret fromMap(Map<String, String> specMap) {
+    Set<String> allowedKeys =
+        new HashSet<>(Arrays.asList("version_name", "name", "project", "version"));
+    Set<String> invalidKeys = new HashSet<>(specMap.keySet());
+    invalidKeys.removeAll(allowedKeys);
+    if (!invalidKeys.isEmpty()) {
+      List<String> sortedInvalid = new ArrayList<>(invalidKeys);
+      Collections.sort(sortedInvalid);
+      throw new IllegalArgumentException(
+          "Invalid secret parameter " + String.join(", ", sortedInvalid));
+    }
+    String versionName = parseVersionName(specMap);
+    return new GcpSecret(versionName);
+  }
+
+  /** Parses the version name from a specification dictionary. */
+  private static String parseVersionName(Map<String, String> specMap) {
+    String versionNameParam = specMap.get("version_name");
+    if (!Strings.isNullOrEmpty(versionNameParam)) {
+      return Preconditions.checkNotNull(
+          versionNameParam, "version_name must contain a valid value for versionName parameter");
+    }
+    String secretId = specMap.get("name");
+    if (Strings.isNullOrEmpty(secretId)) {
+      throw new IllegalArgumentException("Secret name must be specified in secret spec.");
+    }
+    String projectId = resolveGcpProjectId(specMap.get("project"), "secret '" + secretId + "'");
+    String versionId = specMap.getOrDefault("version", "latest");
+    if (Strings.isNullOrEmpty(versionId)) {
+      versionId = "latest";
+    }
+    return String.format("projects/%s/secrets/%s/versions/%s", projectId, secretId, versionId);
+  }
+
+  /**
+   * Resolves the GCP project ID from the provided value, environment variables, or Application
+   * Default Credentials.
+   */
+  static String resolveGcpProjectId(@Nullable String projectId, @Nullable String context) {
+    if (!Strings.isNullOrEmpty(projectId)) {
+      return Preconditions.checkNotNull(projectId);
+    }
+    String envProject = System.getenv("GOOGLE_CLOUD_PROJECT");
+    if (!Strings.isNullOrEmpty(envProject)) {
+      return Preconditions.checkNotNull(envProject);
+    }
+    envProject = System.getenv("GCP_PROJECT");
+    if (!Strings.isNullOrEmpty(envProject)) {
+      return Preconditions.checkNotNull(envProject);
+    }
+    try {
+      Class<?> clazz = Class.forName("com.google.cloud.ServiceOptions");
+      java.lang.reflect.Method method = clazz.getMethod("getDefaultProjectId");
+      @SuppressWarnings("nullness")
+      Object result = method.invoke(null);
+      if (result != null && !Strings.isNullOrEmpty(result.toString())) {
+        return result.toString();
+      }
+    } catch (Throwable e) {
+      LOG.debug("Could not resolve GCP project via ServiceOptions reflection", e);
+    }
+    throw new IllegalArgumentException(
+        String.format(
+            "Could not resolve GCP project ID%s. "
+                + "Please specify 'project' in the secret spec, set GOOGLE_CLOUD_PROJECT environment variable, "
+                + "or configure Application Default Credentials.",
+            context != null ? " for " + context : ""));
+  }
+
   /**
    * Returns the secret as a byte array. Assumes that the current active service account has
    * permissions to read the secret.
@@ -64,4 +148,21 @@ public byte[] getSecretBytes() {
   public String getVersionName() {
     return versionName;
   }
+
+  @Override
+  public boolean equals(@Nullable Object obj) {
+    if (this == obj) {
+      return true;
+    }
+    if (!(obj instanceof GcpSecret)) {
+      return false;
+    }
+    GcpSecret other = (GcpSecret) obj;
+    return Objects.equals(this.versionName, other.versionName);
+  }
+
+  @Override
+  public int hashCode() {
+    return Objects.hash(versionName);
+  }
 }

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

@@ -0,0 +1,57 @@
+/*
+ * 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 java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import org.checkerframework.checker.nullness.qual.Nullable;
+
+/** A {@link Secret} implementation wrapping a raw secret string or byte array directly. */
+public class RawSecret extends Secret {
+  private final byte[] secret;
+
+  public RawSecret(byte[] secret) {
+    this.secret = secret == null ? new byte[0] : secret.clone();
+  }
+
+  public RawSecret(String secret) {
+    this.secret = secret == null ? new byte[0] : secret.getBytes(StandardCharsets.UTF_8);
+  }
+
+  @Override
+  public byte[] getSecretBytes() {
+    return secret.clone();
+  }
+
+  @Override
+  public boolean equals(@Nullable Object obj) {
+    if (this == obj) {
+      return true;
+    }
+    if (!(obj instanceof RawSecret)) {
+      return false;
+    }
+    RawSecret other = (RawSecret) obj;
+    return Arrays.equals(this.secret, other.secret);
+  }
+
+  @Override
+  public int hashCode() {
+    return Arrays.hashCode(secret);
+  }
+}