SQOOP-914: Securing passwords in sqoop 1.x

(Venkatesh Seetharam via Jarek Jarcec Cecho)

src/docs/man/common-args.txt

@@ -39,6 +39,11 @@ Database connection and common options
 --help::
   Print usage instructions
 
+--password-file (file containing the password)::
+  Set authentication password in a file on the users home
+  directory with 400 permissions
+  (Note: This is very secure and a preferred way of entering credentials)
+
 --password (password)::
   Set authentication password
   (Note: This is very insecure. You should use -P instead.)

src/docs/user/common-args.txt

@@ -29,6 +29,8 @@ Argument                                  Description
                                           to use
 +\--hadoop-mapred-home <dir>+             Override $HADOOP_MAPRED_HOME
 +\--help+                                 Print usage instructions
++\--password-file+                        Set path for a file containing the\
+                                          authentication password
 +-P+                                      Read password from console
 +\--password <password>+                  Set authentication password
 +\--username <username>+                  Set authentication username

src/docs/user/connecting.txt

@@ -42,22 +42,41 @@ the full hostname or IP address of the database host that can be seen
 by all your remote nodes.
 
 You might need to authenticate against the database before you can
-access it. You can use the +\--username+ and +\--password+ or +-P+ parameters
-to supply a username and a password to the database. For example:
+access it. You can use the +\--username+ to supply a username to the database.
+Sqoop provides couple of different ways to supply a password,
+secure and non-secure, to the database which is detailed below.
+
+.Secure way of supplying password to the database
+You should save the password in a file on the users home directory with 400
+permissions and specify the path to that file using the *+--password-file+*
+argument, and is the preferred method of entering credentials. Sqoop will
+then read the password from the file and pass it to the MapReduce cluster
+using secure means with out exposing the password in the job configuration.
+The file containing the password can either be on the Local FS or HDFS.
+For example:
+
+----
+$ sqoop import --connect jdbc:mysql://database.example.com/employees \
+    --username venkatesh --passwordFile ${user.home}/.password
+----
+
+Another way of supplying passwords is using the +-P+ argument which will
+read a password from a console prompt.
+
+.Non-secure way of passing password
+
+WARNING: The +\--password+ parameter is insecure, as other users may
+be able to read your password from the command-line arguments via
+the output of programs such as `ps`. The *+-P+* argument is the preferred
+method over using the +\--password+ argument. Credentials may still be
+transferred between nodes of the MapReduce cluster using insecure means.
+For example:
 
 ----
 $ sqoop import --connect jdbc:mysql://database.example.com/employees \
     --username aaron --password 12345
 ----
 
-.Password security
-WARNING: The +\--password+ parameter is insecure, as other users may
-be able to read your password from the command-line arguments via
-the output of programs such as `ps`. The *+-P+* argument will read
-a password from a console prompt, and is the preferred method of
-entering credentials. Credentials may still be transferred between
-nodes of the MapReduce cluster using insecure means.
-
 Sqoop automatically supports several databases, including MySQL.  Connect
 strings beginning with +jdbc:mysql://+ are handled automatically in Sqoop.  (A
 full list of databases with built-in support is provided in the "Supported

src/docs/user/help.txt

@@ -70,6 +70,7 @@ Common arguments:
    --driver <class-name>    Manually specify JDBC driver class to use
    --hadoop-mapred-home <dir>            Override $HADOOP_MAPRED_HOME
    --help                   Print usage instructions
+   --password-file          Set path for file containing authentication password
    -P                       Read password from console
    --password <password>    Set authentication password
    --username <username>    Set authentication username

src/docs/user/tools.txt

@@ -132,6 +132,7 @@ Common arguments:
    --driver <class-name>    Manually specify JDBC driver class to use
    --hadoop-mapred-home <dir>+      Override $HADOOP_MAPRED_HOME
    --help                   Print usage instructions
+   --password-file          Set path for file containing authentication password
    -P                       Read password from console
    --password <password>    Set authentication password
    --username <username>    Set authentication username

src/java/org/apache/sqoop/SqoopOptions.java

@@ -22,6 +22,7 @@
 import com.cloudera.sqoop.SqoopOptions.IncrementalMode;
 import com.cloudera.sqoop.SqoopOptions.UpdateMode;
 import java.io.File;
+import java.io.IOException;
 import java.lang.reflect.Field;
 import java.util.ArrayList;
 import java.util.Arrays;
@@ -38,6 +39,7 @@
 import com.cloudera.sqoop.tool.SqoopTool;
 import com.cloudera.sqoop.util.RandomHash;
 import com.cloudera.sqoop.util.StoredAsProperty;
+import org.apache.sqoop.util.CredentialsUtil;
 import org.apache.sqoop.util.LoggingUtils;
 import org.apache.sqoop.validation.AbsoluteValidationThreshold;
 import org.apache.sqoop.validation.LogOnFailureHandler;
@@ -108,6 +110,10 @@ public String toString() {
   // used. If so, it is stored as 'db.password'.
   private String password;
 
+  // This represents path to a file on ${user.home} containing the password
+  // with 400 permissions so its only readable by user executing the tool
+  @StoredAsProperty("db.password.file") private String passwordFilePath;
+
   @StoredAsProperty("null.string") private String nullStringValue;
   @StoredAsProperty("input.null.string") private String inNullStringValue;
   @StoredAsProperty("null.non-string") private String nullNonStringValue;
@@ -535,13 +541,7 @@ public void loadProperties(Properties props) {
     // Now load properties that were stored with special types, or require
     // additional logic to set.
 
-    if (getBooleanProperty(props, "db.require.password", false)) {
-      // The user's password was stripped out from the metastore.
-      // Require that the user enter it now.
-      setPasswordFromConsole();
-    } else {
-      this.password = props.getProperty("db.password", this.password);
-    }
+    loadPasswordProperty(props);
 
     if (this.jarDirIsAuto) {
       // We memoized a user-specific nonce dir for compilation to the data
@@ -583,6 +583,27 @@ public void loadProperties(Properties props) {
     }
   }
 
+  private void loadPasswordProperty(Properties props) {
+    passwordFilePath = props.getProperty("db.password.file");
+    if (passwordFilePath != null) {
+      try {
+        password = CredentialsUtil.fetchPasswordFromFile(
+                getConf(), passwordFilePath);
+        return; // short-circuit
+      } catch (IOException e) {
+        throw new RuntimeException("Unable to fetch password from file.", e);
+      }
+    }
+
+    if (getBooleanProperty(props, "db.require.password", false)) {
+      // The user's password was stripped out from the metastore.
+      // Require that the user enter it now.
+      setPasswordFromConsole();
+    } else {
+      this.password = props.getProperty("db.password", this.password);
+    }
+  }
+
   /**
    * Return a Properties instance that encapsulates all the "sticky"
    * state of this SqoopOptions that should be written to a metastore
@@ -625,20 +646,7 @@ public Properties writeProperties() {
           iae);
     }
 
-
-    if (this.getConf().getBoolean(
-        METASTORE_PASSWORD_KEY, METASTORE_PASSWORD_DEFAULT)) {
-      // If the user specifies, we may store the password in the metastore.
-      putProperty(props, "db.password", this.password);
-      putProperty(props, "db.require.password", "false");
-    } else if (this.password != null) {
-      // Otherwise, if the user has set a password, we just record
-      // a flag stating that the password will need to be reentered.
-      putProperty(props, "db.require.password", "true");
-    } else {
-      // No password saved or required.
-      putProperty(props, "db.require.password", "false");
-    }
+    writePasswordProperty(props);
 
     putProperty(props, "db.column.list", arrayToList(this.columns));
     setDelimiterProperties(props, "codegen.input.delimiters",
@@ -657,6 +665,27 @@ public Properties writeProperties() {
     return props;
   }
 
+  private void writePasswordProperty(Properties props) {
+    if (getPasswordFilePath() != null) { // short-circuit
+      putProperty(props, "db.password.file", getPasswordFilePath());
+      return;
+    }
+
+    if (this.getConf().getBoolean(
+      METASTORE_PASSWORD_KEY, METASTORE_PASSWORD_DEFAULT)) {
+      // If the user specifies, we may store the password in the metastore.
+      putProperty(props, "db.password", this.password);
+      putProperty(props, "db.require.password", "false");
+    } else if (this.password != null) {
+      // Otherwise, if the user has set a password, we just record
+      // a flag stating that the password will need to be reentered.
+      putProperty(props, "db.require.password", "true");
+    } else {
+      // No password saved or required.
+      putProperty(props, "db.require.password", "false");
+    }
+  }
+
   @Override
   public Object clone() {
     try {
@@ -1037,6 +1066,14 @@ public String getPassword() {
     return password;
   }
 
+  public String getPasswordFilePath() {
+    return passwordFilePath;
+  }
+
+  public void setPasswordFilePath(String passwdFilePath) {
+    this.passwordFilePath = passwdFilePath;
+  }
+
   protected void parseColumnMapping(String mapping,
           Properties output) {
     output.clear();

src/java/org/apache/sqoop/mapreduce/db/DBConfiguration.java

@@ -25,10 +25,14 @@
 import java.util.Map.Entry;
 import java.util.Properties;
 
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
 import org.apache.hadoop.conf.Configuration;
 import org.apache.commons.lang.StringEscapeUtils;
 import org.apache.commons.lang.StringUtils;
 import org.apache.commons.lang.text.StrTokenizer;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.JobConf;
 import org.apache.sqoop.mapreduce.DBWritable;
 
 import com.cloudera.sqoop.mapreduce.db.DBInputFormat.NullDBWritable;
@@ -49,6 +53,9 @@
  */
 public class DBConfiguration {
 
+  public static final Log LOG =
+    LogFactory.getLog(DBConfiguration.class.getName());
+
   /** The JDBC Driver class name. */
   public static final String DRIVER_CLASS_PROPERTY =
     "mapreduce.jdbc.driver.class";
@@ -61,6 +68,8 @@ public class DBConfiguration {
 
   /** Password to access the database. */
   public static final String PASSWORD_PROPERTY = "mapreduce.jdbc.password";
+  private static final Text PASSWORD_SECRET_KEY =
+    new Text(DBConfiguration.PASSWORD_PROPERTY);
 
   /** JDBC connection parameters. */
   public static final String CONNECTION_PARAMS_PROPERTY =
@@ -132,7 +141,7 @@ public static void configureDB(Configuration conf, String driverClass,
       conf.set(USERNAME_PROPERTY, userName);
     }
     if (passwd != null) {
-      conf.set(PASSWORD_PROPERTY, passwd);
+      setPassword((JobConf) conf, passwd);
     }
     if (fetchSize != null) {
       conf.setInt(FETCH_SIZE, fetchSize);
@@ -143,6 +152,13 @@ public static void configureDB(Configuration conf, String driverClass,
     }
   }
 
+  // set the password in the secure credentials object
+  private static void setPassword(JobConf configuration, String password) {
+    LOG.debug("Securing password into job credentials store");
+    configuration.getCredentials().addSecretKey(
+      PASSWORD_SECRET_KEY, password.getBytes());
+  }
+
   /**
    * Sets the DB access related fields in the JobConf.
    * @param job the job
@@ -253,7 +269,7 @@ public Connection getConnection()
     Class.forName(conf.get(DBConfiguration.DRIVER_CLASS_PROPERTY));
 
     String username = conf.get(DBConfiguration.USERNAME_PROPERTY);
-    String password = conf.get(DBConfiguration.PASSWORD_PROPERTY);
+    String password = getPassword((JobConf) conf);
     String connectString = conf.get(DBConfiguration.URL_PROPERTY);
     String connectionParamsStr =
       conf.get(DBConfiguration.CONNECTION_PARAMS_PROPERTY);
@@ -282,6 +298,14 @@ public Connection getConnection()
     return connection;
   }
 
+  // retrieve the password from the credentials object
+  private static String getPassword(JobConf configuration) {
+    LOG.debug("Fetching password from job credentials store");
+    byte[] secret = configuration.getCredentials().getSecretKey(
+      PASSWORD_SECRET_KEY);
+    return secret != null ? new String(secret) : null;
+  }
+
   public Configuration getConf() {
     return conf;
   }

src/java/org/apache/sqoop/tool/BaseSqoopTool.java

@@ -22,6 +22,7 @@
 import java.io.FileInputStream;
 import java.io.IOException;
 import java.io.InputStream;
+import java.io.StringWriter;
 import java.sql.SQLException;
 import java.util.Arrays;
 import java.util.Properties;
@@ -30,8 +31,11 @@
 import org.apache.commons.cli.Option;
 import org.apache.commons.cli.OptionBuilder;
 import org.apache.commons.cli.OptionGroup;
+import org.apache.commons.io.IOUtils;
 import org.apache.commons.logging.Log;
 import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.util.StringUtils;
 
 import com.cloudera.sqoop.ConnFactory;
@@ -71,6 +75,7 @@ public abstract class BaseSqoopTool extends com.cloudera.sqoop.tool.SqoopTool {
   public static final String USERNAME_ARG = "username";
   public static final String PASSWORD_ARG = "password";
   public static final String PASSWORD_PROMPT_ARG = "P";
+  public static final String PASSWORD_PATH_ARG = "password-file";
   public static final String DIRECT_ARG = "direct";
   public static final String BATCH_ARG = "batch";
   public static final String TABLE_ARG = "table";
@@ -382,6 +387,10 @@ protected RelatedOptions getCommonOptions() {
         .hasArg().withDescription("Set authentication password")
         .withLongOpt(PASSWORD_ARG)
         .create());
+    commonOpts.addOption(OptionBuilder.withArgName(PASSWORD_PATH_ARG)
+        .hasArg().withDescription("Set authentication password file path")
+        .withLongOpt(PASSWORD_PATH_ARG)
+        .create());
     commonOpts.addOption(OptionBuilder
         .withDescription("Read password from console")
         .create(PASSWORD_PROMPT_ARG));
@@ -732,6 +741,18 @@ protected void applyCommonOptions(CommandLine in, SqoopOptions out)
       out.setDriverClassName(in.getOptionValue(DRIVER_ARG));
     }
 
+    applyCredentialsOptions(in, out);
+
+    if (in.hasOption(HADOOP_HOME_ARG)) {
+      out.setHadoopMapRedHome(in.getOptionValue(HADOOP_HOME_ARG));
+    }
+    if (in.hasOption(HADOOP_MAPRED_HOME_ARG)) {
+      out.setHadoopMapRedHome(in.getOptionValue(HADOOP_MAPRED_HOME_ARG));
+    }
+  }
+
+  private void applyCredentialsOptions(CommandLine in, SqoopOptions out)
+    throws InvalidOptionsException {
     if (in.hasOption(USERNAME_ARG)) {
       out.setUsername(in.getOptionValue(USERNAME_ARG));
       if (null == out.getPassword()) {
@@ -751,13 +772,56 @@ protected void applyCommonOptions(CommandLine in, SqoopOptions out)
       out.setPasswordFromConsole();
     }
 
-    if (in.hasOption(HADOOP_HOME_ARG)) {
-      out.setHadoopMapRedHome(in.getOptionValue(HADOOP_HOME_ARG));
-    }
-    if (in.hasOption(HADOOP_MAPRED_HOME_ARG)) {
-      out.setHadoopMapRedHome(in.getOptionValue(HADOOP_MAPRED_HOME_ARG));
+    if (in.hasOption(PASSWORD_PATH_ARG)) {
+      if (in.hasOption(PASSWORD_ARG) || in.hasOption(PASSWORD_PROMPT_ARG)) {
+        throw new InvalidOptionsException("Either password or path to a "
+          + "password file must be specified but not both.");
       }
 
+      try {
+        out.setPasswordFilePath(in.getOptionValue(PASSWORD_PATH_ARG));
+        // apply password from file into password in options
+        out.setPassword(fetchPasswordFromFile(out));
+      } catch (IOException ex) {
+        LOG.warn("Failed to load connection parameter file", ex);
+        throw new InvalidOptionsException(
+          "Error while loading connection parameter file: "
+            + ex.getMessage());
+      }
+    }
+  }
+
+  private String fetchPasswordFromFile(SqoopOptions options)
+    throws IOException {
+    String passwordFilePath = options.getPasswordFilePath();
+    if (passwordFilePath == null) {
+      return options.getPassword();
+    }
+
+    LOG.debug("Fetching password from specified path: " + passwordFilePath);
+    FileSystem fs = FileSystem.get(options.getConf());
+    Path path = new Path(passwordFilePath);
+
+    if (!fs.exists(path)) {
+      throw new IOException("The password file does not exist! "
+        + passwordFilePath);
+    }
+
+    if (!fs.isFile(path)) {
+      throw new IOException("The password file cannot be a directory! "
+        + passwordFilePath);
+    }
+
+    InputStream is = fs.open(path);
+    StringWriter writer = new StringWriter();
+    try {
+      IOUtils.copy(is, writer);
+      return writer.toString();
+    } finally {
+      IOUtils.closeQuietly(is);
+      IOUtils.closeQuietly(writer);
+      fs.close();
+    }
   }
 
   protected void applyHiveOptions(CommandLine in, SqoopOptions out)

src/java/org/apache/sqoop/util/CredentialsUtil.java

@@ -0,0 +1,84 @@
+/**
+ * Copyright 2011 The Apache Software Foundation
+ *
+ * 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.sqoop.util;
+
+import com.cloudera.sqoop.SqoopOptions;
+import org.apache.commons.io.IOUtils;
+import org.apache.commons.logging.Log;
+import org.apache.commons.logging.LogFactory;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.StringWriter;
+
+/**
+ * A utility class for fetching passwords from a file.
+ */
+public final class CredentialsUtil {
+
+  public static final Log LOG = LogFactory.getLog(
+    CredentialsUtil.class.getName());
+
+  private CredentialsUtil() {
+  }
+
+  public static String fetchPasswordFromFile(SqoopOptions options)
+    throws IOException {
+    String passwordFilePath = options.getPasswordFilePath();
+    if (passwordFilePath == null) {
+      return options.getPassword();
+    }
+
+    return fetchPasswordFromFile(options.getConf(), passwordFilePath);
+  }
+
+  public static String fetchPasswordFromFile(Configuration conf,
+                                             String passwordFilePath)
+    throws IOException {
+    LOG.debug("Fetching password from specified path: " + passwordFilePath);
+    FileSystem fs = FileSystem.get(conf);
+    Path path = new Path(passwordFilePath);
+
+    if (!fs.exists(path)) {
+      throw new IOException("The password file does not exist! "
+        + passwordFilePath);
+    }
+
+    if (!fs.isFile(path)) {
+      throw new IOException("The password file cannot be a directory! "
+        + passwordFilePath);
+    }
+
+    InputStream is = fs.open(path);
+    StringWriter writer = new StringWriter();
+    try {
+      IOUtils.copy(is, writer);
+      return writer.toString();
+    } finally {
+      IOUtils.closeQuietly(is);
+      IOUtils.closeQuietly(writer);
+      fs.close();
+    }
+  }
+}

src/test/org/apache/sqoop/credentials/TestPassingSecurePassword.java

@@ -0,0 +1,335 @@
+/**
+ * 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.sqoop.credentials;
+
+import com.cloudera.sqoop.SqoopOptions;
+import com.cloudera.sqoop.testutil.BaseSqoopTestCase;
+import com.cloudera.sqoop.testutil.CommonArgs;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapred.JobConf;
+import org.apache.sqoop.mapreduce.db.DBConfiguration;
+import org.apache.sqoop.tool.BaseSqoopTool;
+import org.apache.sqoop.tool.ImportTool;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.sql.Connection;
+import java.util.ArrayList;
+import java.util.Collections;
+import java.util.Properties;
+
+/**
+ * Set of tests for securing passwords.
+ */
+public class TestPassingSecurePassword extends BaseSqoopTestCase {
+
+  @Override
+  public void setUp() {
+    super.setUp();
+    Path warehousePath = new Path(this.getWarehouseDir());
+    try {
+      FileSystem fs = FileSystem.get(getConf());
+      fs.create(warehousePath, true);
+    } catch (IOException e) {
+      System.out.println("Could not create warehouse dir!");
+    }
+  }
+
+  public void testPasswordFilePathInOptionIsEnabled() throws Exception {
+    String passwordFilePath = TEMP_BASE_DIR + ".pwd";
+    createTempFile(passwordFilePath);
+
+    try {
+      ArrayList<String> extraArgs = new ArrayList<String>();
+      extraArgs.add("--username");
+      extraArgs.add("username");
+      extraArgs.add("--password-file");
+      extraArgs.add(passwordFilePath);
+      String[] commonArgs = getCommonArgs(false, extraArgs);
+      ArrayList<String> argsList = new ArrayList<String>();
+      Collections.addAll(argsList, commonArgs);
+      assertTrue("passwordFilePath option missing.",
+        argsList.contains("--password-file"));
+    } catch (Exception e) {
+      fail("passwordPath option is missing.");
+    }
+  }
+
+  public void testPasswordFileDoesNotExist() throws Exception {
+    try {
+      ArrayList<String> extraArgs = new ArrayList<String>();
+      extraArgs.add("--password-file");
+      extraArgs.add(TEMP_BASE_DIR + "unknown");
+      String[] argv = getCommonArgs(false, extraArgs);
+
+      Configuration conf = getConf();
+      SqoopOptions opts = getSqoopOptions(conf);
+      ImportTool importTool = new ImportTool();
+      importTool.parseArguments(argv, conf, opts, true);
+      fail("The password file does not exist! ");
+    } catch (Exception e) {
+      assertTrue(e.getMessage().contains("The password file does not exist!"));
+    }
+  }
+
+  public void testPasswordFileIsADirectory() throws Exception {
+    try {
+      ArrayList<String> extraArgs = new ArrayList<String>();
+      extraArgs.add("--password-file");
+      extraArgs.add(TEMP_BASE_DIR);
+      String[] argv = getCommonArgs(false, extraArgs);
+
+      Configuration conf = getConf();
+      SqoopOptions opts = getSqoopOptions(conf);
+      ImportTool importTool = new ImportTool();
+      importTool.parseArguments(argv, conf, opts, true);
+      fail("The password file cannot be a directory! ");
+    } catch (Exception e) {
+      assertTrue(e.getMessage().contains("The password file cannot "
+        + "be a directory!"));
+    }
+  }
+
+  public void testBothPasswordOptions() throws Exception {
+    String passwordFilePath = TEMP_BASE_DIR + ".pwd";
+    createTempFile(passwordFilePath);
+
+    try {
+      ArrayList<String> extraArgs = new ArrayList<String>();
+      extraArgs.add("--username");
+      extraArgs.add("username");
+      extraArgs.add("--password");
+      extraArgs.add("password");
+      extraArgs.add("--password-file");
+      extraArgs.add(passwordFilePath);
+      String[] argv = getCommonArgs(false, extraArgs);
+
+      Configuration conf = getConf();
+      SqoopOptions in = getSqoopOptions(conf);
+      ImportTool importTool = new ImportTool();
+      SqoopOptions out = importTool.parseArguments(argv, conf, in, true);
+      assertNotNull(out.getPassword());
+      importTool.validateOptions(out);
+      fail("Either password or passwordPath must be specified but not both.");
+    } catch (Exception e) {
+      assertTrue(e.getMessage().contains("Either password or path to a "
+        + "password file must be specified but not both"));
+    }
+  }
+
+  public void testPasswordFilePath() throws Exception {
+    String passwordFilePath = TEMP_BASE_DIR + ".pwd";
+    createTempFile(passwordFilePath);
+    writeToFile(passwordFilePath, "password");
+
+    try {
+      ArrayList<String> extraArgs = new ArrayList<String>();
+      extraArgs.add("--username");
+      extraArgs.add("username");
+      extraArgs.add("--password-file");
+      extraArgs.add(passwordFilePath);
+      String[] commonArgs = getCommonArgs(false, extraArgs);
+
+      Configuration conf = getConf();
+      SqoopOptions in = getSqoopOptions(conf);
+      ImportTool importTool = new ImportTool();
+      SqoopOptions out = importTool.parseArguments(commonArgs, conf, in, true);
+      assertNotNull(out.getPasswordFilePath());
+      assertNotNull(out.getPassword());
+      assertEquals("password", out.getPassword());
+    } catch (Exception e) {
+      fail("passwordPath option is missing.");
+    }
+  }
+
+  public void testPasswordInDBConfiguration() throws Exception {
+    JobConf jobConf = new JobConf(getConf());
+    DBConfiguration.configureDB(jobConf, "org.hsqldb.jdbcDriver",
+      getConnectString(), "username", "password", null, null);
+
+    assertNotNull(jobConf.getCredentials().getSecretKey(
+      new Text(DBConfiguration.PASSWORD_PROPERTY)));
+    assertEquals("password", new String(jobConf.getCredentials().getSecretKey(
+      new Text(DBConfiguration.PASSWORD_PROPERTY))));
+
+    // necessary to wipe the state of previous call to configureDB
+    jobConf = new JobConf();
+    DBConfiguration.configureDB(jobConf, "org.hsqldb.jdbcDriver",
+      getConnectString(), null, null, null, null);
+    DBConfiguration dbConfiguration = new DBConfiguration(jobConf);
+    Connection connection = dbConfiguration.getConnection();
+    assertNotNull(connection);
+  }
+
+  public void testPasswordNotInJobConf() throws Exception {
+    JobConf jobConf = new JobConf(getConf());
+    DBConfiguration.configureDB(jobConf, "org.hsqldb.jdbcDriver",
+      getConnectString(), "username", "password", null, null);
+
+    assertNull(jobConf.get(DBConfiguration.PASSWORD_PROPERTY, null));
+  }
+
+  public void testPasswordInMetastoreWithRecordEnabledAndSecureOption()
+    throws Exception {
+    String passwordFilePath = TEMP_BASE_DIR + ".pwd";
+    createTempFile(passwordFilePath);
+
+    ArrayList<String> extraArgs = new ArrayList<String>();
+    extraArgs.add("--username");
+    extraArgs.add("username");
+    extraArgs.add("--password-file");
+    extraArgs.add(passwordFilePath);
+    String[] argv = getCommonArgs(false, extraArgs);
+
+    Configuration conf = getConf();
+    SqoopOptions in = getSqoopOptions(conf);
+    ImportTool importTool = new ImportTool();
+    SqoopOptions out = importTool.parseArguments(argv, conf, in, true);
+    assertNotNull(out.getPassword());
+
+    // Enable storing passwords in the metastore
+    conf.set(SqoopOptions.METASTORE_PASSWORD_KEY, "true");
+
+    // this is what is used to record password into the metastore
+    Properties propertiesIntoMetastore = out.writeProperties();
+
+    assertNull(propertiesIntoMetastore.getProperty("db.password"));
+    // password-file should NOT be null as it'll be sued to retrieve password
+    assertNotNull(propertiesIntoMetastore.getProperty("db.password.file"));
+
+    // load the saved properties and verify
+    SqoopOptions optionsFromMetastore = new SqoopOptions();
+    optionsFromMetastore.loadProperties(propertiesIntoMetastore);
+    assertNotNull(optionsFromMetastore.getPassword());
+    assertNotNull(optionsFromMetastore.getPasswordFilePath());
+    assertEquals(passwordFilePath, optionsFromMetastore.getPasswordFilePath());
+  }
+
+  public void testPasswordInMetastoreWithRecordDisabledAndSecureOption()
+    throws Exception {
+    String passwordFilePath = TEMP_BASE_DIR + ".pwd";
+    createTempFile(passwordFilePath);
+
+    ArrayList<String> extraArgs = new ArrayList<String>();
+    extraArgs.add("--username");
+    extraArgs.add("username");
+    extraArgs.add("--password-file");
+    extraArgs.add(passwordFilePath);
+    String[] argv = getCommonArgs(false, extraArgs);
+
+    Configuration conf = getConf();
+    SqoopOptions in = getSqoopOptions(conf);
+    ImportTool importTool = new ImportTool();
+    SqoopOptions out = importTool.parseArguments(argv, conf, in, true);
+    assertNotNull(out.getPassword());
+
+    // Enable storing passwords in the metastore
+    conf.set(SqoopOptions.METASTORE_PASSWORD_KEY, "false");
+
+    // this is what is used to record password into the metastore
+    Properties propertiesIntoMetastore = out.writeProperties();
+
+    assertNull(propertiesIntoMetastore.getProperty("db.password"));
+    assertNotNull(propertiesIntoMetastore.getProperty("db.password.file"));
+
+    // load the saved properties and verify
+    SqoopOptions optionsFromMetastore = new SqoopOptions();
+    optionsFromMetastore.loadProperties(propertiesIntoMetastore);
+    assertNotNull(optionsFromMetastore.getPassword());
+    assertNotNull(optionsFromMetastore.getPasswordFilePath());
+    assertEquals(passwordFilePath, optionsFromMetastore.getPasswordFilePath());
+  }
+
+  public void testPasswordInMetastoreWithRecordEnabledAndNonSecureOption()
+    throws Exception {
+    ArrayList<String> extraArgs = new ArrayList<String>();
+    extraArgs.add("--username");
+    extraArgs.add("username");
+    extraArgs.add("--password");
+    extraArgs.add("password");
+    String[] argv = getCommonArgs(false, extraArgs);
+
+    Configuration conf = getConf();
+    SqoopOptions in = getSqoopOptions(conf);
+    ImportTool importTool = new ImportTool();
+    SqoopOptions out = importTool.parseArguments(argv, conf, in, true);
+    assertNotNull(out.getPassword());
+
+    // Enable storing passwords in the metastore
+    conf.set(SqoopOptions.METASTORE_PASSWORD_KEY, "true");
+
+    // this is what is used to record password into the metastore
+    Properties propertiesIntoMetastore = out.writeProperties();
+
+    assertNotNull(propertiesIntoMetastore.getProperty("db.password"));
+    assertNull(propertiesIntoMetastore.getProperty("db.password.file"));
+
+    // load the saved properties and verify
+    SqoopOptions optionsFromMetastore = new SqoopOptions();
+    optionsFromMetastore.loadProperties(propertiesIntoMetastore);
+    assertNotNull(optionsFromMetastore.getPassword());
+    assertNull(optionsFromMetastore.getPasswordFilePath());
+  }
+
+  private String[] getCommonArgs(boolean includeHadoopFlags,
+                                 ArrayList<String> extraArgs) {
+    ArrayList<String> args = new ArrayList<String>();
+
+    if (includeHadoopFlags) {
+      CommonArgs.addHadoopFlags(args);
+    }
+
+    args.add("--table");
+    args.add(getTableName());
+    args.add("--warehouse-dir");
+    args.add(getWarehouseDir());
+    args.add("--connect");
+    args.add(getConnectString());
+    args.add("--as-textfile");
+    args.add("--num-mappers");
+    args.add("2");
+
+    args.addAll(extraArgs);
+
+    return args.toArray(new String[0]);
+  }
+
+  private void createTempFile(String filePath) throws IOException {
+    File pwdFile = new File(filePath);
+    pwdFile.createNewFile();
+  }
+
+  private void writeToFile(String filePath, String contents)
+    throws IOException {
+    File pwdFile = new File(filePath);
+    FileOutputStream fos = null;
+    try {
+      fos = new FileOutputStream(pwdFile);
+      fos.write(contents.getBytes());
+    } finally {
+      if (fos != null) {
+        fos.close();
+      }
+    }
+  }
+}