<a id="elasticsearch-connector-security"></a>

# Elasticsearch Connector with Security

Complete the following instructions to configure the Kafka Connect Elasticsearch connector with security. These instructions are based on the Elasticsearch document [Encrypting HTTP Client communications](https://www.elastic.co/guide/en/elasticsearch/reference/6.6/configuring-tls.html#tls-http).

Prerequisites:
: * Confluent Platform version 4.1 (or later) using the bundled Elasticsearch connector.
  * Elasticsearch with the X-Pack plugin. This plugin is pre-installed in Elasticsearch versions 6.3 or later.

## Step 1: Download and extract the Elastic archive

Enter the following commands:

```none
wget https://artifacts.elastic.co/downloads/elasticsearch/elasticsearch-6.6.0.tar.gz
```

```none
tar xzvf elasticsearch-6.6.0.tar.gz
```

```none
cd elasticsearch-6.6.0
```

## Step 2: Generate Certificates

Elasticsearch provides a certificate-generation utility named `bin/elasticsearch-certutil`. By default, this utility creates self-signed certificates. Use the commands below to generate your own certificates. This allows you to simulate a production-like environment.

1. Make a certificate working directory in the Elasticsearch `config` directory:
   ```none
   mkdir config/certs
   ```

   ```none
   cd config/certs
   ```
2. Generate the Certificate Authority (make sure your FQDN is `localhost`):
   ```none
   openssl req -new -x509 -keyout cacert.key -out cacert.pem -days 666
   ```
3. Generate a client certificate:
   ```none
   openssl genrsa -out client1.key 2048
   ```
4. Generate a certificate signing request:
   ```none
   openssl req -new -key client1.key -out client1.csr
   ```
5. Sign the request with the CA:
   ```none
   openssl x509 -req -in client1.csr -CA cacert.pem -CAkey cacert.key \
   -CAcreateserial -out client1.crt -days 1825 -sha256
   ```
6. Repeat the previous steps for the next client:
   ```none
   openssl genrsa -out client2.key 2048
   ```

   ```none
   openssl req -new -key client2.key -out client2.csr
   ```

   ```none
   openssl x509 -req -in client2.csr -CA cacert.pem -CAkey cacert.key \
   -CAcreateserial -out client2.crt -days 1825 -sha256
   ```
7. Package the connector keys as JKS:
   ```none
   openssl pkcs12 -export -out bundle.p12 -in client2.crt -inkey client2.key
   ```

   ```none
   keytool -keystore truststore.jks -import -file cacert.pem -alias cacert
   ```

   ```none
   keytool -destkeystore keystore.jks -importkeystore -srckeystore bundle.p12 -srcstoretype PKCS12
   ```

## Step 3: Configure Elasticsearch

1. Return to the main Elasticsearch directory:
   ```none
   cd ../..
   ```
2. Update the Elasticsearch configuration file:
   ```none
   cat <<EOF >> config/elasticsearch.yml
   xpack.security.enabled: true
   xpack.security.http.ssl.enabled: true
   xpack.security.http.ssl.client_authentication: required
   xpack.security.http.ssl.key:  certs/client1.key
   xpack.security.http.ssl.certificate: certs/client1.crt
   xpack.security.http.ssl.certificate_authorities: [ "certs/cacert.pem" ]
   EOF
   ```
3. Set the passwords:
   ```none
   bin/elasticsearch-keystore add xpack.security.http.ssl.secure_key_passphrase
   ```
4. Run Elasticsearch:
   ```none
   bin/elasticsearch
   ```
5. Test the connection:
   ```none
   curl --key config/certs/client2.key --cert config/certs/client2.crt \
   --cacert config/certs/cacert.pem https://localhost:9200
   ```

## Step 4: Configure the Connector

Prerequisites
: - [Confluent Platform](/platform/current/installation/index.html)
  - [Confluent CLI](https://docs.confluent.io/confluent-cli/current/installing.html) (requires separate installation)

1. Open a new terminal and change your current directory to `<path-to-confluent>`.
2. Save the configuration file as `elastic.properties` in `etc/kafka-connect-elasticsearch` and add the certificate paths.
   ```none
   cat <<EOF > etc/kafka-connect-elasticsearch/elastic-secure.properties
   name=elasticsearch-sink
   connector.class=io.confluent.connect.elasticsearch.ElasticsearchSinkConnector
   tasks.max=1
   topics=test-elasticsearch-sink
   key.ignore=true
   connection.url=https://localhost:9200
   type.name=kafka-connect

   elastic.security.protocol=SSL
   elastic.https.ssl.keystore.location=/home/directory/elasticsearch-6.6.0/config/certs/keystore.jks
   elastic.https.ssl.keystore.password=asdfasdf
   elastic.https.ssl.key.password=asdfasdf
   elastic.https.ssl.keystore.type=JKS
   elastic.https.ssl.truststore.location=/home/directory/elasticsearch-6.6.0/config/certs/truststore.jks
   elastic.https.ssl.truststore.password=asdfasdf
   elastic.https.ssl.truststore.type=JKS
   elastic.https.ssl.protocol=TLS
   EOF
   ```
3. Start Connect and load the connector:
   ```none
   bin/confluent local services connect start
   ```

   ```none
   bin/confluent local services connect connector load elasticssl --config etc/kafka-connect-elasticsearch/elastic-secure.properties
   ```

## Step 5: Test the System

1. Enter the following command:
   ```none
   bin/kafka-avro-console-producer --broker-list localhost:9092 --topic test-elasticsearch-sink \
   --property value.schema='{"type":"record","name":"myrecord","fields":[{"name":"f1","type":"string"}]}'
   {"f1": "secret1"}
   {"f1": "secret2"}
   ```
2. Query Elasticsearch:
   ```none
   curl --key config/certs/client2.key --cert config/certs/client2.crt /
   --cacert config/certs/cacert.pem 'https://localhost:9200/test-elasticsearch-sink/_search?pretty'
   ```