[AMORO-4359][AMS] Support AWS Secrets Manager as a ConfigShade

LICENSE-binary

@@ -328,6 +328,7 @@ software.amazon.awssdk:protocol-core
 software.amazon.awssdk:regions
 software.amazon.awssdk:s3
 software.amazon.awssdk:sdk-core
+software.amazon.awssdk:secretsmanager
 software.amazon.awssdk:sts
 software.amazon.awssdk:third-party-jackson-core
 software.amazon.awssdk:url-connection-client

amoro-ams/pom.xml

@@ -336,6 +336,22 @@
             <artifactId>url-connection-client</artifactId>
         </dependency>
 
+        <!-- AWS Secrets Manager config shade (uses url-connection-client above) -->
+        <dependency>
+            <groupId>software.amazon.awssdk</groupId>
+            <artifactId>secretsmanager</artifactId>
+            <exclusions>
+                <exclusion>
+                    <groupId>software.amazon.awssdk</groupId>
+                    <artifactId>netty-nio-client</artifactId>
+                </exclusion>
+                <exclusion>
+                    <groupId>software.amazon.awssdk</groupId>
+                    <artifactId>apache-client</artifactId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+
         <dependency>
             <groupId>software.amazon.awssdk</groupId>
             <artifactId>s3-transfer-manager</artifactId>

amoro-ams/src/main/java/org/apache/amoro/server/config/shade/AwsSecretsManagerClient.java

@@ -0,0 +1,42 @@
+/*
+ * 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.amoro.server.config.shade;
+
+/**
+ * A minimal abstraction over AWS Secrets Manager. It only exposes a single method for fetching a
+ * secret string, which makes it easy to stub in unit tests.
+ *
+ * <p>Implementations are expected to cache within the process — AMS only calls {@link
+ * #getSecretString(String)} during startup, but the same secret may be referenced in several places
+ * (for example, the username/password of the same DB living in the same JSON).
+ */
+interface AwsSecretsManagerClient extends AutoCloseable {
+
+  /**
+   * Fetches the {@code SecretString} content of the secret.
+   *
+   * @param secretId the secret name or ARN
+   * @return the plaintext of the secret (either a plain string or JSON)
+   * @throws RuntimeException if the fetch fails; the caller decides whether to fail-fast
+   */
+  String getSecretString(String secretId);
+
+  @Override
+  void close();
+}

amoro-ams/src/main/java/org/apache/amoro/server/config/shade/AwsSecretsManagerConfigShade.java

@@ -0,0 +1,186 @@
+/*
+ * 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.amoro.server.config.shade;
+
+import org.apache.amoro.config.Configurations;
+import org.apache.amoro.config.shade.ConfigShade;
+import org.apache.amoro.shade.jackson2.com.fasterxml.jackson.databind.JsonNode;
+import org.apache.amoro.shade.jackson2.com.fasterxml.jackson.databind.ObjectMapper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.IOException;
+import java.util.Objects;
+import java.util.function.Function;
+
+/**
+ * Delegates AMS's sensitive configs (db.username / db.password, etc.) to AWS Secrets Manager
+ * through Amoro's {@link ConfigShade} SPI.
+ *
+ * <p>Usage (in {@code $AMORO_HOME/conf/config.yaml}):
+ *
+ * <pre>{@code
+ * ams:
+ *   shade:
+ *     identifier: aws-sm
+ *     sensitive-keywords: database.username;database.password
+ *   database:
+ *     # both point to the same JSON secret; '#<field>' selects the value out of it
+ *     username: arn:aws:secretsmanager:ap-northeast-1:123456789012:secret:prod-amoro-db-mOhyOp#db.username
+ *     password: arn:aws:secretsmanager:ap-northeast-1:123456789012:secret:prod-amoro-db-mOhyOp#db.password
+ * }</pre>
+ *
+ * <p>The trailing {@code -mOhyOp} is the six random characters AWS appends to every secret ARN; use
+ * the full ARN as shown. The optional {@code #<field>} suffix selects one field when the secret
+ * value is JSON, e.g. {@code {"db.username": "amoro", "db.password": "..."}} — the field name is
+ * whatever key you stored, taken as-is (dots are part of the key, not a nested path). Omit the
+ * suffix when the secret value is the plaintext itself.
+ *
+ * <p>No explicit region configuration is required — it is resolved from the ARN. All secrets shaded
+ * for one AMS instance are expected to live in the same region, so a single client is created
+ * (lazily, memoized within the process); a reference resolving to a different region is treated as
+ * a misconfiguration and fails fast.
+ *
+ * <p>AMS startup flow: {@code AmoroServiceContainer.initServiceConfig} → {@code
+ * ConfigShadeUtils.decryptConfig} → find this SPI instance with identifier=aws-sm → call {@link
+ * #decrypt(String)} for each sensitive key.
+ *
+ * <p>Failure strategy: always fail-fast (malformed ARN, fetch failure, missing JSON field) so that
+ * AMS startup blows up immediately, instead of continuing with a wrong password — errors that
+ * surface after the DB receives a wrong password are far from the root cause.
+ */
+public class AwsSecretsManagerConfigShade implements ConfigShade {
+
+  private static final Logger LOG = LoggerFactory.getLogger(AwsSecretsManagerConfigShade.class);
+
+  public static final String IDENTIFIER = "aws-sm";
+
+  private final ObjectMapper jsonMapper;
+
+  /**
+   * A single client, lazily initialized on the first decrypt() call. All secrets shaded for one AMS
+   * instance (admin-password, database.password, ...) live in the same region, so one client is
+   * enough. Guarded by {@code this} for the double-checked lazy init; the region it was bound to is
+   * validated against every subsequent reference so that a misconfigured cross-region ARN fails
+   * fast instead of being fetched with the wrong client.
+   */
+  private volatile AwsSecretsManagerClient client;
+
+  private volatile String boundRegion;
+
+  /**
+   * The factory that creates a client; {@link DefaultAwsSecretsManagerClient#create} for
+   * production, a mock injected in unit tests.
+   */
+  private final Function<String, AwsSecretsManagerClient> clientFactory;
+
+  /** No-arg SPI constructor. */
+  public AwsSecretsManagerConfigShade() {
+    this(new ObjectMapper(), DefaultAwsSecretsManagerClient::create);
+  }
+
+  /** For injecting a mock client factory in unit tests. */
+  AwsSecretsManagerConfigShade(
+      ObjectMapper jsonMapper, Function<String, AwsSecretsManagerClient> clientFactory) {
+    this.jsonMapper = Objects.requireNonNull(jsonMapper, "jsonMapper");
+    this.clientFactory = Objects.requireNonNull(clientFactory, "clientFactory");
+  }
+
+  @Override
+  public String getIdentifier() {
+    return IDENTIFIER;
+  }
+
+  @Override
+  public void initialize(Configurations serviceConfig) {
+    // No SPI-level config to read — region is resolved per ARN; credentials use the AWS default
+    // provider chain.
+    LOG.info("AwsSecretsManagerConfigShade initialized (region resolved per-secret from ARN)");
+  }
+
+  @Override
+  public String decrypt(String content) {
+    SecretReference ref = SecretReference.parse(content);
+    String raw = getOrCreateClient(ref.region()).getSecretString(ref.secretArn());
+    return ref.jsonKey().map(key -> extractJsonField(raw, key, ref.secretArn())).orElse(raw);
+  }
+
+  /**
+   * Returns the shared client, creating it on the first call. All shaded secrets are expected to
+   * live in a single region; if a later reference resolves to a different region it is almost
+   * certainly a misconfiguration, so we fail fast rather than fetch it with the wrong client.
+   */
+  private AwsSecretsManagerClient getOrCreateClient(String region) {
+    AwsSecretsManagerClient existing = client;
+    if (existing == null) {
+      synchronized (this) {
+        existing = client;
+        if (existing == null) {
+          existing = clientFactory.apply(region);
+          boundRegion = region;
+          client = existing;
+        }
+      }
+    }
+    if (!Objects.equals(region, boundRegion)) {
+      throw new IllegalStateException(
+          "all AWS Secrets Manager references must be in the same region; expected '"
+              + boundRegion
+              + "' but got '"
+              + region
+              + "'");
+    }
+    return existing;
+  }
+
+  private String extractJsonField(String json, String key, String secretArn) {
+    JsonNode root;
+    try {
+      root = jsonMapper.readTree(json);
+    } catch (IOException e) {
+      // Do not log the raw JSON; it may contain other sensitive fields.
+      throw new IllegalStateException(
+          "secret '" + secretArn + "' is not valid JSON; cannot extract field '" + key + "'", e);
+    }
+    if (root == null || !root.isObject()) {
+      throw new IllegalStateException(
+          "secret '" + secretArn + "' is not a JSON object; cannot extract field '" + key + "'");
+    }
+    JsonNode value = root.get(key);
+    if (value == null || value.isNull()) {
+      throw new IllegalStateException("secret '" + secretArn + "' has no field '" + key + "'");
+    }
+    // A config value must be a scalar. For an object/array node asText() silently returns "", which
+    // would then be handed to the DB as e.g. an empty password — the far-from-root-cause failure
+    // this class fails fast to avoid. So reject non-scalar values explicitly.
+    if (!value.isValueNode()) {
+      throw new IllegalStateException(
+          "secret '"
+              + secretArn
+              + "' field '"
+              + key
+              + "' must be a scalar value, but is a "
+              + value.getNodeType()
+              + "; a structured JSON value cannot be used as a config value");
+    }
+    // asText() returns strings as-is and converts numbers/booleans to strings — a good fit for the
+    // password scenario.
+    return value.asText();
+  }
+}

amoro-ams/src/main/java/org/apache/amoro/server/config/shade/DefaultAwsSecretsManagerClient.java

@@ -0,0 +1,112 @@
+/*
+ * 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.amoro.server.config.shade;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.auth.credentials.DefaultCredentialsProvider;
+import software.amazon.awssdk.http.urlconnection.UrlConnectionHttpClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.secretsmanager.SecretsManagerClient;
+import software.amazon.awssdk.services.secretsmanager.model.GetSecretValueRequest;
+import software.amazon.awssdk.services.secretsmanager.model.GetSecretValueResponse;
+
+import java.util.Objects;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.ConcurrentMap;
+
+/**
+ * The default {@link AwsSecretsManagerClient} implementation, bound to a single region.
+ *
+ * <ul>
+ *   <li>Credentials: {@link DefaultCredentialsProvider} — tries env / Java system props / Web
+ *       Identity (EKS IRSA) / ECS / EC2 InstanceProfile / {@code ~/.aws/credentials} in order.
+ *   <li>Region: extracted from the ARN by the caller and passed in at construction time.
+ *   <li>HTTP: {@link UrlConnectionHttpClient} — no Netty dependency, minimizing the shade artifact.
+ *   <li>Cache: an in-process {@link ConcurrentHashMap}; each secret is fetched only once (username
+ *       and password usually live in the same JSON and are referenced twice).
+ * </ul>
+ *
+ * <p>On failure it throws {@link IllegalStateException} to make AMS startup fail-fast, avoiding
+ * starting the service with a wrong password.
+ */
+final class DefaultAwsSecretsManagerClient implements AwsSecretsManagerClient {
+
+  private static final Logger LOG = LoggerFactory.getLogger(DefaultAwsSecretsManagerClient.class);
+
+  private final SecretsManagerClient delegate;
+  private final String region;
+  private final ConcurrentMap<String, String> cache = new ConcurrentHashMap<>();
+
+  private DefaultAwsSecretsManagerClient(SecretsManagerClient delegate, String region) {
+    this.delegate = Objects.requireNonNull(delegate, "delegate");
+    this.region = region;
+  }
+
+  static DefaultAwsSecretsManagerClient create(String region) {
+    Objects.requireNonNull(region, "region");
+    SecretsManagerClient sdkClient =
+        SecretsManagerClient.builder()
+            .region(Region.of(region))
+            .credentialsProvider(DefaultCredentialsProvider.create())
+            .httpClient(UrlConnectionHttpClient.create())
+            .build();
+    LOG.info("AWS Secrets Manager client initialized (region={})", region);
+    return new DefaultAwsSecretsManagerClient(sdkClient, region);
+  }
+
+  @Override
+  public String getSecretString(String secretId) {
+    return cache.computeIfAbsent(secretId, this::fetch);
+  }
+
+  private String fetch(String secretId) {
+    try {
+      GetSecretValueResponse response =
+          delegate.getSecretValue(GetSecretValueRequest.builder().secretId(secretId).build());
+      String value = response.secretString();
+      if (value == null) {
+        // Binary secrets are rarely used for passwords; fail clearly if one is encountered.
+        throw new IllegalStateException(
+            "secret '" + secretId + "' has no SecretString (binary secrets are not supported)");
+      }
+      LOG.info("Fetched secret '{}' from AWS Secrets Manager (region={})", secretId, region);
+      return value;
+    } catch (RuntimeException e) {
+      // Mask the secret value itself, but keep secretId / region for operational troubleshooting.
+      throw new IllegalStateException(
+          "Failed to fetch secret '"
+              + secretId
+              + "' from AWS Secrets Manager (region="
+              + region
+              + "): "
+              + e.getMessage(),
+          e);
+    }
+  }
+
+  @Override
+  public void close() {
+    try {
+      delegate.close();
+    } catch (RuntimeException e) {
+      LOG.warn("Error closing SecretsManagerClient (region={})", region, e);
+    }
+  }
+}