/*
 * Copyright (C) 2014-2015 LinkedIn Corp. All rights reserved.
 *
 * Licensed 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.
 */

package gobblin.password;

import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.jasypt.util.text.StrongTextEncryptor;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Charsets;
import com.google.common.base.Optional;
import com.google.common.base.Preconditions;
import com.google.common.cache.CacheBuilder;
import com.google.common.cache.CacheLoader;
import com.google.common.cache.LoadingCache;
import com.google.common.io.Closer;
import com.google.common.io.LineReader;

import gobblin.configuration.ConfigurationKeys;
import gobblin.configuration.State;


/**
 * A class for managing password encryption and decryption. To encrypt or decrypt a password, a master password
 * should be provided which is used as encryption or decryption key.
 *
 * @author ziliu
 */
public class PasswordManager {

  private static final Logger LOG = LoggerFactory.getLogger(PasswordManager.class);

  private static final long CACHE_SIZE = 100;
  private static final long CACHE_EXPIRATION_MIN = 10;
  private static final Pattern PASSWORD_PATTERN = Pattern.compile("ENC\\((.*)\\)");

  private static final LoadingCache<Optional<String>, PasswordManager> CACHED_INSTANCES =
      CacheBuilder.newBuilder().maximumSize(CACHE_SIZE).expireAfterAccess(CACHE_EXPIRATION_MIN, TimeUnit.MINUTES)
          .build(new CacheLoader<Optional<String>, PasswordManager>() {
            public PasswordManager load(Optional<String> masterPassword) {
              return new PasswordManager(masterPassword);
            }
          });

  private final Optional<StrongTextEncryptor> encryptor;

  private PasswordManager(Optional<String> masterPassword) {
    if (masterPassword.isPresent()) {
      this.encryptor = Optional.of(new StrongTextEncryptor());
      this.encryptor.get().setPassword(masterPassword.get());
    } else {
      this.encryptor = Optional.absent();
    }
  }

  /**
   * Get an instance with no master password, which cannot encrypt or decrypt passwords.
   */
  public static PasswordManager getInstance() {
    try {
      Optional<String> absent = Optional.absent();
      return CACHED_INSTANCES.get(absent);
    } catch (ExecutionException e) {
      throw new RuntimeException("Unable to get an instance of PasswordManager", e);
    }
  }

  /**
   * Get an instance. The location of the master password file is provided via "encrypt.key.loc".
   */
  public static PasswordManager getInstance(State state) {
    try {
      return CACHED_INSTANCES.get(getMasterPassword(state));
    } catch (ExecutionException e) {
      throw new RuntimeException("Unable to get an instance of PasswordManager", e);
    }
  }

  /**
   * Get an instance. The location of the master password file is provided via "encrypt.key.loc".
   */
  public static PasswordManager getInstance(Properties props) {
    try {
      return CACHED_INSTANCES.get(getMasterPassword(new State(props)));
    } catch (ExecutionException e) {
      throw new RuntimeException("Unable to get an instance of PasswordManager", e);
    }
  }

  /**
   * Get an instance. The master password file is given by masterPwdLoc.
   */
  public static PasswordManager getInstance(Path masterPwdLoc) {
    try {
      return CACHED_INSTANCES.get(getMasterPassword(masterPwdLoc));
    } catch (ExecutionException e) {
      throw new RuntimeException("Unable to get an instance of PasswordManager", e);
    }
  }

  /**
   * Encrypt a password. A master password must have been provided in the constructor.
   * @param plain A plain password to be encrypted.
   * @return The encrypted password.
   */
  public String encryptPassword(String plain) {
    Preconditions.checkArgument(this.encryptor.isPresent(),
        "A master password needs to be provided for encrypting passwords.");

    try {
      return this.encryptor.get().encrypt(plain);
    } catch (Exception e) {
      throw new RuntimeException("Failed to encrypt password", e);
    }
  }

  /**
   * Decrypt an encrypted password. A master password must have been provided in the constructor.
   * @param encrypted An encrypted password.
   * @return The decrypted password.
   */
  public String decryptPassword(String encrypted) {
    Preconditions.checkArgument(this.encryptor.isPresent(),
        "A master password needs to be provided for decrypting passwords.");

    try {
      return this.encryptor.get().decrypt(encrypted);
    } catch (Exception e) {
      throw new RuntimeException("Failed to decrypt password " + encrypted, e);
    }
  }

  /**
   * Decrypt a password if it is an encrypted password (in the form of ENC(.*)), and a master password has been
   * provided in the constructor. Otherwise, return the password as is.
   */
  public String readPassword(String password) {
    if (!this.encryptor.isPresent()) {
      return password;
    }
    Matcher matcher = PASSWORD_PATTERN.matcher(password);
    if (matcher.matches()) {
      return this.decryptPassword(password);
    } else {
      return password;
    }
  }

  private static Optional<String> getMasterPassword(State state) {
    if (!state.contains(ConfigurationKeys.ENCRYPT_KEY_LOC)) {
      LOG.warn(String.format("Property %s not set. Cannot decrypt any encrypted password.",
          ConfigurationKeys.ENCRYPT_KEY_LOC));
      return Optional.absent();
    }
    try {
      return getMasterPassword(new Path(state.getProp(ConfigurationKeys.ENCRYPT_KEY_LOC)));
    } catch (Exception e) {
      throw new RuntimeException(
          "Failed to obtain master password from " + state.getProp(ConfigurationKeys.ENCRYPT_KEY_LOC), e);
    }
  }

  @SuppressWarnings("deprecation")
  private static Optional<String> getMasterPassword(Path masterPwdLoc) {
    Closer closer = Closer.create();
    try {
      FileSystem fs = masterPwdLoc.getFileSystem(new Configuration());
      if (!fs.exists(masterPwdLoc) || fs.getFileStatus(masterPwdLoc).isDir()) {
        LOG.warn(masterPwdLoc + " does not exist or is not a file. Cannot decrypt any encrypted password.");
        return Optional.absent();
      }
      InputStream in = closer.register(fs.open(masterPwdLoc));
      return Optional.of(new LineReader(new InputStreamReader(in, Charsets.UTF_8)).readLine());
    } catch (IOException e) {
      throw new RuntimeException("Failed to obtain master password from " + masterPwdLoc, e);
    } finally {
      try {
        closer.close();
      } catch (IOException e) {
        throw new RuntimeException("Failed to close inputstream for " + masterPwdLoc, e);
      }
    }
  }
}

@@ -5755,15 +5755,12 @@
 her.
-matches
+find
 ()) 
@@ -5795,24 +5795,32 @@
 assword(
-password
+matcher.group(1)
 );%0A    %7D

Fix bug in PasswordManager