<commit_before>/*
 * Copyright 2017 StreamSets 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 com.streamsets.pipeline.lib.elasticsearch;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.google.gson.JsonPrimitive;
import com.google.gson.JsonSyntaxException;
import com.streamsets.pipeline.api.Stage;
import com.streamsets.pipeline.api.StageException;
import com.streamsets.pipeline.lib.aws.AwsUtil;
import com.streamsets.pipeline.stage.config.elasticsearch.ElasticsearchConfig;
import com.streamsets.pipeline.stage.config.elasticsearch.Errors;
import com.streamsets.pipeline.stage.config.elasticsearch.Groups;
import com.streamsets.pipeline.stage.config.elasticsearch.SecurityConfig;
import com.streamsets.pipeline.stage.config.elasticsearch.SecurityMode;
import com.streamsets.pipeline.stage.lib.aws.AwsRegion;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.lang.StringUtils;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHost;
import org.apache.http.HttpRequestInterceptor;
import org.apache.http.HttpStatus;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.message.BasicHeader;
import org.apache.http.ssl.SSLContexts;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.ResponseException;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.sniff.ElasticsearchHostsSniffer;
import org.elasticsearch.client.sniff.HostsSniffer;
import org.elasticsearch.client.sniff.Sniffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.SSLContext;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.CertificateException;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Scanner;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static com.google.gson.FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES;
import static com.streamsets.pipeline.stage.config.elasticsearch.Errors.ELASTICSEARCH_09;
import static com.streamsets.pipeline.stage.config.elasticsearch.Errors.ELASTICSEARCH_46;
import static com.streamsets.pipeline.stage.config.elasticsearch.Errors.ELASTICSEARCH_47;
import static com.streamsets.pipeline.stage.config.elasticsearch.Errors.ELASTICSEARCH_48;
import static com.streamsets.pipeline.stage.config.elasticsearch.SecurityMode.BASIC;
import static org.apache.http.entity.ContentType.APPLICATION_JSON;

public class ElasticsearchStageDelegate {
  private static final Logger LOG = LoggerFactory.getLogger(ElasticsearchStageDelegate.class);
  private static final Pattern URI_PATTERN = Pattern.compile("\\S+:(\\d+)");
  private static final String AWS_SERVICE_NAME = "es";
  private static final JsonParser JSON_PARSER = new JsonParser();
  private static final Gson GSON = new GsonBuilder().setFieldNamingPolicy(LOWER_CASE_WITH_UNDERSCORES).create();
  private static final String VALID_PROPERTY_NAME = "valid";
  private static final String QUERY_PROPERTY_NAME = "query";
  private static final String VALIDATE_QUERY_PATH = "/_validate/query";
  private static final String HTTP_URIS_CONFIG_NAME = "httpUris";
  private static final String QUERY_CONFIG_NAME = "query";
  private static final String SECURITY_CONFIG_NAME = "securityConfig";
  private static final String USER_CONFIG_NAME = "securityUser";
  private static final String USE_SECURITY_CONFIG_NAME = "useSecurity";
  private static final String ACCESS_KEY_ID_CONFIG_NAME = "awsAccessKeyId";

  private final Stage.Context context;
  private final ElasticsearchConfig conf;
  private RestClient restClient;
  private Sniffer sniffer;
  private String version = "not-known";
  private int majorVersion = -1;

  public ElasticsearchStageDelegate(Stage.Context context, ElasticsearchConfig conf) {
    this.context = context;
    this.conf = conf;
  }

  public List<Stage.ConfigIssue> init(String prefix, List<Stage.ConfigIssue> issues) {
    if (conf.httpUris.isEmpty()) {
      issues.add(
          context.createConfigIssue(
              Groups.ELASTIC_SEARCH.name(),
              prefix + ".httpUris",
              Errors.ELASTICSEARCH_06
          )
      );
    } else {
      for (String uri : conf.httpUris) {
        validateUri(uri, issues, prefix + ".httpUris");
      }
    }

    String securityUser = null;
    try {
      securityUser = conf.securityConfig.securityUser.get();
    } catch (StageException e) {
       issues.add(context.createConfigIssue(
           Groups.SECURITY.name(),
           prefix + "." + SecurityConfig.NAME + ".securityUser",
           Errors.ELASTICSEARCH_32,
           e.getMessage(),
           e
      ));
    }

    String securityPassword = null;
    try {
      securityPassword = conf.securityConfig.securityPassword.get();
    } catch (StageException e) {
      issues.add(context.createConfigIssue(
          Groups.SECURITY.name(),
          prefix + "." + SecurityConfig.NAME + ".securityPassword",
          Errors.ELASTICSEARCH_38,
          e.getMessage(),
          e
      ));
    }

    if (conf.useSecurity && SecurityMode.BASIC.equals(conf.securityConfig.securityMode)) {
      if (securityUser == null || securityPassword == null) {
        issues.add(
            context.createConfigIssue(
                Groups.SECURITY.name(),
                prefix + "." + SecurityConfig.NAME + ".securityUser",
                Errors.ELASTICSEARCH_40
            )
        );
      } else {
        if (securityUser.isEmpty()) {
          issues.add(
              context.createConfigIssue(
                  Groups.SECURITY.name(),
                  prefix + "." + SecurityConfig.NAME + ".securityUser",
                  Errors.ELASTICSEARCH_20
              )
          );
        } else if (!securityUser.contains(":") && securityPassword.isEmpty()) {
          issues.add(<commit_msg>SDC-15855. Don't resolve username and password if AWS security is used<commit_after>/*
 * Copyright 2017 StreamSets 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 com.streamsets.pipeline.lib.elasticsearch;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import com.google.gson.JsonElement;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.google.gson.JsonPrimitive;
import com.google.gson.JsonSyntaxException;
import com.streamsets.pipeline.api.Stage;
import com.streamsets.pipeline.api.StageException;
import com.streamsets.pipeline.lib.aws.AwsUtil;
import com.streamsets.pipeline.stage.config.elasticsearch.ElasticsearchConfig;
import com.streamsets.pipeline.stage.config.elasticsearch.Errors;
import com.streamsets.pipeline.stage.config.elasticsearch.Groups;
import com.streamsets.pipeline.stage.config.elasticsearch.SecurityConfig;
import com.streamsets.pipeline.stage.config.elasticsearch.SecurityMode;
import com.streamsets.pipeline.stage.lib.aws.AwsRegion;
import org.apache.commons.codec.binary.Base64;
import org.apache.commons.lang.StringUtils;
import org.apache.http.Header;
import org.apache.http.HttpEntity;
import org.apache.http.HttpHost;
import org.apache.http.HttpRequestInterceptor;
import org.apache.http.HttpStatus;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.nio.client.HttpAsyncClientBuilder;
import org.apache.http.message.BasicHeader;
import org.apache.http.ssl.SSLContexts;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.ResponseException;
import org.elasticsearch.client.RestClient;
import org.elasticsearch.client.RestClientBuilder;
import org.elasticsearch.client.sniff.ElasticsearchHostsSniffer;
import org.elasticsearch.client.sniff.HostsSniffer;
import org.elasticsearch.client.sniff.Sniffer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import javax.net.ssl.SSLContext;
import java.io.IOException;
import java.io.InputStream;
import java.nio.file.Files;
import java.nio.file.Path;
import java.nio.file.Paths;
import java.security.KeyManagementException;
import java.security.KeyStore;
import java.security.KeyStoreException;
import java.security.NoSuchAlgorithmException;
import java.security.cert.CertificateException;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Scanner;
import java.util.regex.Matcher;
import java.util.regex.Pattern;
import java.util.stream.Collectors;

import static com.google.gson.FieldNamingPolicy.LOWER_CASE_WITH_UNDERSCORES;
import static com.streamsets.pipeline.stage.config.elasticsearch.Errors.ELASTICSEARCH_09;
import static com.streamsets.pipeline.stage.config.elasticsearch.Errors.ELASTICSEARCH_46;
import static com.streamsets.pipeline.stage.config.elasticsearch.Errors.ELASTICSEARCH_47;
import static com.streamsets.pipeline.stage.config.elasticsearch.Errors.ELASTICSEARCH_48;
import static com.streamsets.pipeline.stage.config.elasticsearch.SecurityMode.BASIC;
import static org.apache.http.entity.ContentType.APPLICATION_JSON;

public class ElasticsearchStageDelegate {
  private static final Logger LOG = LoggerFactory.getLogger(ElasticsearchStageDelegate.class);
  private static final Pattern URI_PATTERN = Pattern.compile("\\S+:(\\d+)");
  private static final String AWS_SERVICE_NAME = "es";
  private static final JsonParser JSON_PARSER = new JsonParser();
  private static final Gson GSON = new GsonBuilder().setFieldNamingPolicy(LOWER_CASE_WITH_UNDERSCORES).create();
  private static final String VALID_PROPERTY_NAME = "valid";
  private static final String QUERY_PROPERTY_NAME = "query";
  private static final String VALIDATE_QUERY_PATH = "/_validate/query";
  private static final String HTTP_URIS_CONFIG_NAME = "httpUris";
  private static final String QUERY_CONFIG_NAME = "query";
  private static final String SECURITY_CONFIG_NAME = "securityConfig";
  private static final String USER_CONFIG_NAME = "securityUser";
  private static final String USE_SECURITY_CONFIG_NAME = "useSecurity";
  private static final String ACCESS_KEY_ID_CONFIG_NAME = "awsAccessKeyId";

  private final Stage.Context context;
  private final ElasticsearchConfig conf;
  private RestClient restClient;
  private Sniffer sniffer;
  private String version = "not-known";
  private int majorVersion = -1;

  public ElasticsearchStageDelegate(Stage.Context context, ElasticsearchConfig conf) {
    this.context = context;
    this.conf = conf;
  }

  public List<Stage.ConfigIssue> init(String prefix, List<Stage.ConfigIssue> issues) {
    if (conf.httpUris.isEmpty()) {
      issues.add(
          context.createConfigIssue(
              Groups.ELASTIC_SEARCH.name(),
              prefix + ".httpUris",
              Errors.ELASTICSEARCH_06
          )
      );
    } else {
      for (String uri : conf.httpUris) {
        validateUri(uri, issues, prefix + ".httpUris");
      }
    }

    String securityUser = null;
    String securityPassword = null;

    if (conf.useSecurity && SecurityMode.BASIC.equals(conf.securityConfig.securityMode)) {
      try {
        securityUser = conf.securityConfig.securityUser.get();
      } catch (StageException e) {
         issues.add(context.createConfigIssue(
             Groups.SECURITY.name(),
             prefix + "." + SecurityConfig.NAME + ".securityUser",
             Errors.ELASTICSEARCH_32,
             e.getMessage(),
             e
        ));
      }
      try {
        securityPassword = conf.securityConfig.securityPassword.get();
      } catch (StageException e) {
        issues.add(context.createConfigIssue(
            Groups.SECURITY.name(),
            prefix + "." + SecurityConfig.NAME + ".securityPassword",
            Errors.ELASTICSEARCH_38,
            e.getMessage(),
            e
        ));
      }
      if (securityUser == null || securityPassword == null) {
        issues.add(
            context.createConfigIssue(
                Groups.SECURITY.name(),
                prefix + "." + SecurityConfig.NAME + ".securityUser",
                Errors.ELASTICSEARCH_40
            )
        );
      } else {
        if (securityUser.isEmpty()) {
          issues.add(
              context.createConfigIssue(
                  Groups.SECURITY.name(),
                  prefix + "." + SecurityConfig.NAME + ".securityUser",
                  Errors.ELASTICSEARCH_20
              )
          );
        } else if (!securityUser.contains(":") && securityPassword.isEmpty()) {
          issues.add(
{'fraction_non_alphanumeric': 0.09761126119721314, 'fraction_numerical': 0.004763258922223802, 'mean_word_length': 3.7912125340599454, 'pattern_counts': {'":': 2, '<': 7, '<?xml version=': 0, '>': 7, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1700862', 'n_tokens_mistral': 3990, 'n_tokens_neox': 3975, 'n_words': 859}