Added decrypt

Refactoring to reduce duplicate code

Some cleanup

Addressed comments

docs/SensitiveRecordDecrypt-transform.md

@@ -0,0 +1,30 @@
+# Cloud Data Loss Prevention (DLP) Decrypt
+
+Additional Charges
+-----------
+This plugin uses Google's Data Loss Prevention APIs which charge the user depending 
+on the volume of data **analyzed** (not transformed). More details on the exact 
+costs can be found [here](https://cloud.google.com/dlp/pricing#content-pricing). 
+
+Permissions
+-----------
+In order for this plugin to function, it requires permissions to access the Data Loss Prevention APIs. These permissions
+granted through the service account that is provided in the plugin configuration. If the service account path is set to 
+`auto-detect` then it will use a service account with the name `service-<project-number>@gcp-sa-datafusion.iam.gserviceaccount.com`.
+
+The `DLP Administrator` role must be granted to the service account to allow this plugin to access the DLP APIs.
+
+Description
+-----------
+This plugin decrypts sensitive data that was encrypted by DLP using a reversible encryption transform, such as `Format 
+Preserving Encryption`. The plugin works by reversing the encryption specified in the config. Therefore, you must provide 
+the same configuration properties that were used to encrypt the data. In other words, the configuration in this plugin 
+and the DLP Redaction plugin must be identical for the decrypt to function correctly.
+
+
+Metrics
+-----------
+This plugin records three metrics:
+* `dlp.requests.count`: Total number of requests sent to Data Loss Prevention API
+* `dlp.requests.success`: Number of requests that were successfully processed by Data Loss Prevention API
+* `dlp.requests.fail`: Number of requests that failed

src/main/java/io/cdap/plugin/dlp/DLPTransformPluginConfig.java

@@ -0,0 +1,165 @@
+/*
+ * Copyright © 2020 Cask Data, Inc.
+ *
+ * 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. See the
+ * License for the specific language governing permissions and limitations under
+ * the License.
+ */
+
+package io.cdap.plugin.dlp;
+
+import com.google.gson.Gson;
+import com.google.gson.GsonBuilder;
+import io.cdap.cdap.api.annotation.Description;
+import io.cdap.cdap.api.annotation.Macro;
+import io.cdap.cdap.api.data.schema.Schema;
+import io.cdap.cdap.etl.api.FailureCollector;
+import io.cdap.plugin.dlp.configs.DlpFieldTransformationConfig;
+import io.cdap.plugin.dlp.configs.DlpFieldTransformationConfigCodec;
+import io.cdap.plugin.dlp.configs.ErrorConfig;
+import io.cdap.plugin.gcp.common.GCPConfig;
+
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+import javax.annotation.Nullable;
+
+/**
+ * Common config used by the DLP Redact and Decrypt plugins
+ */
+public class DLPTransformPluginConfig extends GCPConfig {
+  public static final String FIELDS_TO_TRANSFORM = "fieldsToTransform";
+  @Macro
+  protected String fieldsToTransform;
+
+  @Description("Enabling this option will allow you to define a custom DLP Inspection Template to use for matching "
+    + "during the transform.")
+  protected Boolean customTemplateEnabled;
+
+  @Description("ID of the DLP Inspection template")
+  @Macro
+  @Nullable
+  protected String templateId;
+
+  private static final Gson GSON = new GsonBuilder()
+    .registerTypeAdapter(DlpFieldTransformationConfig.class, new DlpFieldTransformationConfigCodec())
+    .create();
+
+  public List<DlpFieldTransformationConfig> parseTransformations() throws Exception {
+    String[] values = GSON.fromJson(fieldsToTransform, String[].class);
+    List<DlpFieldTransformationConfig> transformationConfigs = new ArrayList<>();
+    for (String value : values) {
+      transformationConfigs.add(GSON.fromJson(value, DlpFieldTransformationConfig.class));
+    }
+    return transformationConfigs;
+  }
+
+  /**
+   * Get the set of fields that are being transformed or are required for transforms to work. This is used to limit
+   * the payload size to DLP endpoints, the transform will only send the values of the required fields.
+   *
+   * @return Set of field names
+   */
+  public Set<String> getRequiredFields() throws Exception {
+    return parseTransformations().stream()
+      .map(DlpFieldTransformationConfig::getRequiredFields)
+      .flatMap(Collection::stream)
+      .collect(Collectors.toSet());
+  }
+
+  public void validate(FailureCollector collector, Schema inputSchema) {
+    if (customTemplateEnabled) {
+      if (!containsMacro("templateId") && templateId == null) {
+        collector.addFailure("Must specify template ID in order to use custom template", "")
+          .withConfigProperty("templateId");
+      }
+    }
+
+    if (fieldsToTransform != null) {
+      try {
+        List<DlpFieldTransformationConfig> transformationConfigs = parseTransformations();
+        HashMap<String, String> transforms = new HashMap<>();
+        Boolean firstTransformUsedCustomTemplate = null;
+        Boolean anyTransformUsedCustomTemplate = false;
+        for (DlpFieldTransformationConfig config : transformationConfigs) {
+          ErrorConfig errorConfig = config.getErrorConfig("");
+
+          //Checking that custom template is defined if it is selected in one of the transforms
+          List<String> filters = Arrays.asList(config.getFilters());
+          if (!customTemplateEnabled && filters.contains("NONE")) {
+            collector.addFailure(String.format("This transform depends on custom template that was not defined.",
+                                               config.getTransform(), String.join(", ", config.getFields())),
+                                 "Enable the custom template option and provide the name of it.")
+              .withConfigElement(FIELDS_TO_TRANSFORM, GSON.toJson(errorConfig));
+          }
+          //Validate the config for the transform
+          config.validate(collector, inputSchema, FIELDS_TO_TRANSFORM);
+
+          //Check that custom template and built-in types are not mixed
+          anyTransformUsedCustomTemplate = anyTransformUsedCustomTemplate || filters.contains("NONE");
+          if (firstTransformUsedCustomTemplate == null) {
+            firstTransformUsedCustomTemplate = filters.contains("NONE");
+          } else {
+            if (filters.contains("NONE") != firstTransformUsedCustomTemplate) {
+              errorConfig.setTransformPropertyId("filters");
+              collector.addFailure("Cannot use custom templates and built-in filters in the same plugin instance.",
+                                   "All transforms must use custom templates or built-in filters, not a "
+                                     + "combination of both.")
+                .withConfigElement(FIELDS_TO_TRANSFORM, GSON.toJson(errorConfig));
+            }
+          }
+
+          // Make sure the combination of field, transform and filter are unique
+          for (String field : config.getFields()) {
+            for (String filter : config.getFilterDisplayNames()) {
+              String transformKey = String.format("%s:%s", field, filter);
+              if (transforms.containsKey(transformKey)) {
+
+                String errorMessage;
+                if (transforms.get(transformKey).equalsIgnoreCase(config.getTransform())) {
+                  errorMessage = String.format(
+                    "Combination of transform, filter and field must be unique. Found multiple definitions for '%s' "
+                      + "transform on '%s' with filter '%s'", config.getTransform(), field, filter);
+                } else {
+                  errorMessage = String.format(
+                    "Only one transform can be defined per field and filter combination. Found conflicting transforms"
+                      + " '%s' and '%s'",
+                    transforms.get(transformKey), config.getTransform());
+                }
+                errorConfig.setTransformPropertyId("");
+                collector.addFailure(errorMessage, "")
+                  .withConfigElement(FIELDS_TO_TRANSFORM, GSON.toJson(errorConfig));
+              } else {
+                transforms.put(transformKey, config.getTransform());
+              }
+            }
+          }
+        }
+
+        // If the user has a custom template enabled but doesnt use it in any of the transforms
+        if (!anyTransformUsedCustomTemplate && this.customTemplateEnabled) {
+          collector.addFailure("Custom template is enabled but no transforms use a custom template.",
+                               "Please define a transform that uses the custom template or disable the custom "
+                                 + "template.")
+            .withConfigProperty("customTemplateEnabled");
+        }
+      } catch (Exception e) {
+        collector.addFailure(String.format("Error while parsing transforms: %s", e.getMessage()), "")
+          .withConfigProperty(FIELDS_TO_TRANSFORM);
+      }
+    }
+  }
+}