[AMBARI-22886] Infra Manager: store s3 credentials in Hadoop credential store

ambari-infra/ambari-infra-manager/src/main/java/org/apache/ambari/infra/conf/security/CompositePasswordStore.java

@@ -0,0 +1,39 @@
+/*
+ * 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.ambari.infra.conf.security;
+
+import java.util.Optional;
+
+public class CompositePasswordStore implements PasswordStore {
+  private PasswordStore[] passwordStores;
+
+  public CompositePasswordStore(PasswordStore... passwordStores) {
+    this.passwordStores = passwordStores;
+  }
+
+  @Override
+  public Optional<String> getPassword(String propertyName) {
+    for (PasswordStore passwordStore : passwordStores) {
+      Optional<String> optionalPassword = passwordStore.getPassword(propertyName);
+      if (optionalPassword.isPresent())
+        return optionalPassword;
+    }
+    return Optional.empty();
+  }
+}

ambari-infra/ambari-infra-manager/src/main/java/org/apache/ambari/infra/conf/security/HadoopCredentialStore.java

@@ -0,0 +1,55 @@
+/*
+ * 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.ambari.infra.conf.security;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Optional;
+
+import static org.apache.commons.lang.StringUtils.isBlank;
+import static org.apache.commons.lang3.ArrayUtils.isNotEmpty;
+
+public class HadoopCredentialStore implements PasswordStore {
+  private static final Logger LOG = LoggerFactory.getLogger(InfraManagerSecurityConfig.class);
+  public static final String CREDENTIAL_STORE_PROVIDER_PATH_PROPERTY = "hadoop.security.credential.provider.path";
+
+  private final String credentialStoreProviderPath;
+
+  public HadoopCredentialStore(String credentialStoreProviderPath) {
+    this.credentialStoreProviderPath = credentialStoreProviderPath;
+  }
+
+  @Override
+  public Optional<String> getPassword(String propertyName) {
+    try {
+      if (isBlank(credentialStoreProviderPath)) {
+        return Optional.empty();
+      }
+
+      org.apache.hadoop.conf.Configuration config = new org.apache.hadoop.conf.Configuration();
+      config.set(CREDENTIAL_STORE_PROVIDER_PATH_PROPERTY, credentialStoreProviderPath);
+      char[] passwordChars = config.getPassword(propertyName);
+      return (isNotEmpty(passwordChars)) ? Optional.of(new String(passwordChars)) : Optional.empty();
+    } catch (Exception e) {
+      LOG.warn("Could not load password {} from credential store.", propertyName);
+      return Optional.empty();
+    }
+  }
+}

ambari-infra/ambari-infra-manager/src/main/java/org/apache/ambari/infra/conf/security/InfraManagerSecurityConfig.java

@@ -0,0 +1,38 @@
+/*
+ * 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.ambari.infra.conf.security;
+
+import org.springframework.beans.factory.annotation.Value;
+import org.springframework.context.annotation.Bean;
+import org.springframework.context.annotation.Configuration;
+
+import static org.apache.ambari.infra.conf.security.HadoopCredentialStore.CREDENTIAL_STORE_PROVIDER_PATH_PROPERTY;
+
+@Configuration
+public class InfraManagerSecurityConfig {
+
+  @Value("${"+ CREDENTIAL_STORE_PROVIDER_PATH_PROPERTY + ":}")
+  private String credentialStoreProviderPath;
+
+
+  @Bean
+  public PasswordStore passwords() {
+    return new CompositePasswordStore(new HadoopCredentialStore(credentialStoreProviderPath), new SecurityEnvironment());
+  }
+}

ambari-infra/ambari-infra-manager/src/main/java/org/apache/ambari/infra/conf/security/PasswordStore.java

@@ -0,0 +1,25 @@
+/*
+ * 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.ambari.infra.conf.security;
+
+import java.util.Optional;
+
+public interface PasswordStore {
+  Optional<String> getPassword(String propertyName);
+}

ambari-infra/ambari-infra-manager/src/main/java/org/apache/ambari/infra/conf/security/SecurityEnvironment.java

@@ -0,0 +1,28 @@
+/*
+ * 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.ambari.infra.conf.security;
+
+import java.util.Optional;
+
+public class SecurityEnvironment implements PasswordStore {
+  @Override
+  public Optional<String> getPassword(String propertyName) {
+    return Optional.ofNullable(System.getenv(propertyName));
+  }
+}

ambari-infra/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/AbstractJobsConfiguration.java

@@ -50,16 +50,20 @@ public void registerJobs() {
     if (propertyMap == null)
       return;
 
-    for (String jobName : propertyMap.keySet())
-      propertyMap.get(jobName).validate(jobName);
-
     propertyMap.keySet().stream()
             .filter(key -> propertyMap.get(key).isEnabled())
             .forEach(jobName -> {
-              LOG.info("Registering job {}", jobName);
-              JobBuilder jobBuilder = jobs.get(jobName).listener(new JobsPropertyMap<>(propertyMap));
-              Job job = buildJob(jobBuilder);
-              jobRegistryBeanPostProcessor.postProcessAfterInitialization(job, jobName);
+              try {
+                propertyMap.get(jobName).validate(jobName);
+                LOG.info("Registering job {}", jobName);
+                JobBuilder jobBuilder = jobs.get(jobName).listener(new JobsPropertyMap<>(propertyMap));
+                Job job = buildJob(jobBuilder);
+                jobRegistryBeanPostProcessor.postProcessAfterInitialization(job, jobName);
+              }
+              catch (Exception e) {
+                LOG.warn("Unable to register job " + jobName, e);
+                propertyMap.get(jobName).setEnabled(false);
+              }
             });
   }
 

ambari-infra/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/JobProperties.java

@@ -68,7 +68,7 @@ public void validate(String jobName) {
       validate();
     }
     catch (Exception ex) {
-      throw new JobConfigurationException(String.format("Configuration of job %s is invalid!", jobName), ex);
+      throw new JobConfigurationException(String.format("Configuration of job %s is invalid: %s!", jobName, ex.getMessage()), ex);
     }
   }
 

ambari-infra/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/DocumentArchivingConfiguration.java

@@ -19,6 +19,7 @@
 package org.apache.ambari.infra.job.archive;
 
 import org.apache.ambari.infra.conf.InfraManagerDataConfig;
+import org.apache.ambari.infra.conf.security.PasswordStore;
 import org.apache.ambari.infra.job.AbstractJobsConfiguration;
 import org.apache.ambari.infra.job.JobContextRepository;
 import org.apache.ambari.infra.job.JobScheduler;
@@ -86,13 +87,16 @@ public DocumentExporter documentExporter(DocumentItemReader documentItemReader,
                                            InfraManagerDataConfig infraManagerDataConfig,
                                            @Value("#{jobParameters[end]}") String intervalEnd,
                                            DocumentWiper documentWiper,
-                                           JobContextRepository jobContextRepository) {
+                                           JobContextRepository jobContextRepository,
+                                           PasswordStore passwordStore) {
 
     File baseDir = new File(infraManagerDataConfig.getDataFolder(), "exporting");
     CompositeFileAction fileAction = new CompositeFileAction(new TarGzCompressor());
     switch (properties.getDestination()) {
       case S3:
-        fileAction.add(new S3Uploader(properties.s3Properties().orElseThrow(() -> new IllegalStateException("S3 properties are not provided!"))));
+        fileAction.add(new S3Uploader(
+                properties.s3Properties().orElseThrow(() -> new IllegalStateException("S3 properties are not provided!")),
+                passwordStore));
         break;
       case HDFS:
         org.apache.hadoop.conf.Configuration conf = new org.apache.hadoop.conf.Configuration();

ambari-infra/ambari-infra-manager/src/main/java/org/apache/ambari/infra/job/archive/DocumentArchivingProperties.java

@@ -19,23 +19,14 @@
 package org.apache.ambari.infra.job.archive;
 
 import org.apache.ambari.infra.job.JobProperties;
-import org.apache.commons.csv.CSVParser;
-import org.apache.commons.csv.CSVRecord;
 import org.springframework.batch.core.JobParameters;
 
-import java.io.FileReader;
-import java.io.IOException;
-import java.io.UncheckedIOException;
-import java.util.Iterator;
-import java.util.Map;
 import java.util.Optional;
-import java.util.function.Supplier;
 
 import static java.util.Objects.requireNonNull;
 import static org.apache.ambari.infra.job.archive.ExportDestination.HDFS;
 import static org.apache.ambari.infra.job.archive.ExportDestination.LOCAL;
 import static org.apache.ambari.infra.job.archive.ExportDestination.S3;
-import static org.apache.commons.csv.CSVFormat.DEFAULT;
 import static org.apache.commons.lang.StringUtils.isBlank;
 
 public class DocumentArchivingProperties extends JobProperties<DocumentArchivingProperties> {
@@ -50,47 +41,12 @@ public class DocumentArchivingProperties extends JobProperties<DocumentArchiving
   private String s3KeyPrefix;
   private String s3BucketName;
   private String s3Endpoint;
-  private transient Supplier<Optional<S3Properties>> s3Properties;
 
   private String hdfsEndpoint;
   private String hdfsDestinationDirectory;
 
   public DocumentArchivingProperties() {
     super(DocumentArchivingProperties.class);
-    s3Properties = this::loadS3Properties;
-  }
-
-  private Optional<S3Properties> loadS3Properties() {
-    if (isBlank(s3BucketName))
-      return Optional.empty();
-
-    String accessKey = System.getenv("AWS_ACCESS_KEY_ID");
-    String secretKey = System.getenv("AWS_SECRET_ACCESS_KEY");
-
-    if (isBlank(accessKey) || isBlank(secretKey)) {
-      if (isBlank(s3AccessFile))
-        return Optional.empty();
-      try (CSVParser csvParser = CSVParser.parse(new FileReader(s3AccessFile), DEFAULT.withHeader("Access key ID", "Secret access key"))) {
-        Iterator<CSVRecord> iterator = csvParser.iterator();
-        if (!iterator.hasNext()) {
-          return Optional.empty();
-        }
-
-        CSVRecord record = csvParser.iterator().next();
-        Map<String, Integer> header = csvParser.getHeaderMap();
-        accessKey = record.get(header.get("Access key ID"));
-        secretKey = record.get(header.get("Secret access key"));
-      } catch (IOException e) {
-        throw new UncheckedIOException(e);
-      }
-    }
-
-    return Optional.of(new S3Properties(
-            accessKey,
-            secretKey,
-            s3KeyPrefix,
-            s3BucketName,
-            s3Endpoint));
   }
 
   public int getReadBlockSize() {
@@ -155,7 +111,6 @@ public String getS3AccessFile() {
 
   public void setS3AccessFile(String s3AccessFile) {
     this.s3AccessFile = s3AccessFile;
-    s3Properties = this::loadS3Properties;
   }
 
   public String getS3KeyPrefix() {
@@ -183,7 +138,14 @@ public void setS3Endpoint(String s3Endpoint) {
   }
 
   public Optional<S3Properties> s3Properties() {
-    return s3Properties.get();
+    if (isBlank(s3BucketName))
+      return Optional.empty();
+
+    return Optional.of(new S3Properties(
+            s3AccessFile,
+            s3KeyPrefix,
+            s3BucketName,
+            s3Endpoint));
   }
 
   public String getHdfsEndpoint() {