Add a support for standard SSL configurations, proxy and basic auth configs.

.ci/run.sh

@@ -0,0 +1,33 @@
+#!/bin/bash
+# This is intended to be run inside the docker container as the command of the docker-compose.
+
+env
+
+set -ex
+
+if [[ "$INTEGRATION" != "true" ]]; then
+  bundle exec rake test
+else
+  # Define the Kafka:Confluent version pairs
+  VERSIONS=(
+    "4.1.0:8.0.0"
+  )
+
+  for pair in "${VERSIONS[@]}"; do
+    KAFKA_VERSION="${pair%%:*}"
+    CONFLUENT_VERSION="${pair##*:}"
+
+    echo "=================================================="
+    echo " Testing with Kafka $KAFKA_VERSION / Confluent $CONFLUENT_VERSION"
+    echo "=================================================="
+
+    export KAFKA_VERSION
+    export CONFLUENT_VERSION
+
+    cd spec/integration && ./kafka_test_setup.sh && cd ../..
+    bundle exec rspec -fd --tag integration
+    cd spec/integration && ./kafka_test_teardown.sh && cd ../..
+  done
+fi
+
+

.travis.yml

@@ -1,2 +1,11 @@
 import:
-- logstash-plugins/.ci:travis/travis.yml@1.x
+- logstash-plugins/.ci:travis/travis.yml@1.x
+
+jobs:
+  include:
+    - stage: "Integration Tests"
+      env: INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=8.current
+    - env: INTEGRATION=true LOG_LEVEL=info ELASTIC_STACK_VERSION=9.current
+    - env: INTEGRATION=true SNAPSHOT=true LOG_LEVEL=info ELASTIC_STACK_VERSION=8.current
+    - env: INTEGRATION=true SNAPSHOT=true LOG_LEVEL=info ELASTIC_STACK_VERSION=9.current
+    - env: INTEGRATION=true SNAPSHOT=true LOG_LEVEL=info ELASTIC_STACK_VERSION=main

CHANGELOG.md

@@ -1,3 +1,17 @@
+## 3.5.0
+  - Add SSL/TLS support for HTTPS schema registry connections
+    - Add `ssl_enabled` option to enable/disable SSL
+    - Add `ssl_certificate` and `ssl_key` options for PEM-based client authentication (unencrypted keys only)
+    - Add `ssl_certificate_authorities` option for PEM-based server certificate validation
+    - Add `ssl_verification_mode` option to control SSL verification (full, none)
+    - Add `ssl_cipher_suites` option to configure cipher suites
+    - Add `ssl_supported_protocols` option to configure TLS protocol versions (TLSv1.1, TLSv1.2, TLSv1.3)
+    - Add `ssl_truststore_path` and `ssl_truststore_password` options for server certificate validation (JKS/PKCS12)
+    - Add `ssl_keystore_path` and `ssl_keystore_password` options for mutual TLS authentication (JKS/PKCS12)
+    - Add `ssl_truststore_type` and `ssl_keystore_type` options (JKS or PKCS12)
+  - Add HTTP proxy support with `proxy` option
+  - Add HTTP basic authentication support with `username` and `password` options
+
 ## 3.4.1
   - Fixes `(Errno::ENOENT) No such file or directory` error [#43](https://github.com/logstash-plugins/logstash-codec-avro/pull/43)
 

docs/index.asciidoc

@@ -83,9 +83,25 @@ output {
 |Setting |Input type|Required
 | <<plugins-{type}s-{plugin}-ecs_compatibility>> | <<string,string>>|No
 | <<plugins-{type}s-{plugin}-encoding>> | <<string,string>>, one of `["binary", "base64"]`|No
+| <<plugins-{type}s-{plugin}-password>> |<<password,password>>|No
+| <<plugins-{type}s-{plugin}-proxy>> |<<uri,uri>>|No
 | <<plugins-{type}s-{plugin}-schema_uri>> |<<string,string>>|Yes
+| <<plugins-{type}s-{plugin}-ssl_certificate>> |<<path,path>>|No
+| <<plugins-{type}s-{plugin}-ssl_certificate_authorities>> |list of <<path,path>>|No
+| <<plugins-{type}s-{plugin}-ssl_cipher_suites>> |<<array,array>>|No
+| <<plugins-{type}s-{plugin}-ssl_enabled>> |<<boolean,boolean>>|No
+| <<plugins-{type}s-{plugin}-ssl_key>> |<<path,path>>|No
+| <<plugins-{type}s-{plugin}-ssl_keystore_password>> |<<password,password>>|No
+| <<plugins-{type}s-{plugin}-ssl_keystore_path>> |<<path,path>>|No
+| <<plugins-{type}s-{plugin}-ssl_keystore_type>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-ssl_supported_protocols>> |<<array,array>>|No
+| <<plugins-{type}s-{plugin}-ssl_truststore_password>> |<<password,password>>|No
+| <<plugins-{type}s-{plugin}-ssl_truststore_path>> |<<path,path>>|No
+| <<plugins-{type}s-{plugin}-ssl_truststore_type>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-ssl_verification_mode>> |<<string,string>>|No
 | <<plugins-{type}s-{plugin}-tag_on_failure>> |<<boolean,boolean>>|No
 | <<plugins-{type}s-{plugin}-target>> |<<string,string>>|No
+| <<plugins-{type}s-{plugin}-username>> |<<string,string>>|No
 |=======================================================================
 
 &nbsp;
@@ -112,6 +128,23 @@ Use `base64` (default) to indicate that this codec sends or expects to receive b
 Set this option to `binary` to indicate that this codec sends or expects to receive binary Avro data.
 
 
+[id="plugins-{type}s-{plugin}-password"]
+===== `password`
+
+* Value type is <<password,password>>
+* There is no default value for this setting.
+
+Password for HTTP basic authentication when fetching remote schemas.
+Used together with `username`.
+
+[id="plugins-{type}s-{plugin}-proxy"]
+===== `proxy`
+
+* Value type is <<uri,uri>>
+* There is no default value for this setting.
+
+The address of a forward HTTP proxy to use when contacting a remote schema registry.
+
 [id="plugins-{type}s-{plugin}-schema_uri"]
 ===== `schema_uri` 
 
@@ -134,6 +167,172 @@ example:
 
 tag events with `_avroparsefailure` when decode fails
 
+[id="plugins-{type}s-{plugin}-ssl_certificate"]
+===== `ssl_certificate`
+
+* Value type is <<path,path>>
+* There is no default value for this setting.
+
+Path to PEM encoded certificate file for client authentication (mutual TLS).
+You may use this setting or <<plugins-{type}s-{plugin}-ssl_keystore_path>>, but not both simultaneously.
+
+*Example*
+[source,ruby]
+----------------------------------
+ssl_certificate => "/path/to/client.crt"
+----------------------------------
+
+[id="plugins-{type}s-{plugin}-ssl_certificate_authorities"]
+===== `ssl_certificate_authorities`
+
+* Value type is a list of <<path,path>>
+* There is no default value for this setting.
+
+Path to PEM encoded CA certificate file(s) for server verification.
+This is an alternative to using <<plugins-{type}s-{plugin}-ssl_truststore_path>>.
+You may use this setting or <<plugins-{type}s-{plugin}-ssl_truststore_path>>, but not both simultaneously.
+
+*Example*
+[source,ruby]
+----------------------------------
+ssl_certificate_authorities => ["/path/to/ca.crt"]
+----------------------------------
+
+[id="plugins-{type}s-{plugin}-ssl_cipher_suites"]
+===== `ssl_cipher_suites`
+
+* Value type is <<array,array>>
+* There is no default value for this setting.
+
+The list of cipher suites to use, listed by priorities.
+Supported cipher suites vary depending on which version of Java is used.
+
+[id="plugins-{type}s-{plugin}-ssl_key"]
+===== `ssl_key`
+
+* Value type is <<path,path>>
+* There is no default value for this setting.
+
+Path to PEM encoded private key file for client authentication.
+Must be used together with <<plugins-{type}s-{plugin}-ssl_certificate>>.
+The private key must be unencrypted (passphrase-protected keys are not supported).
+
+*Example*
+[source,ruby]
+----------------------------------
+ssl_key => "/path/to/client.key"
+----------------------------------
+
+[id="plugins-{type}s-{plugin}-ssl_enabled"]
+===== `ssl_enabled`
+
+* Value type is <<boolean,boolean>>
+* There is no default value for this setting.
+
+Enable SSL/TLS secured communication to remote schema registry.
+When using HTTPS schema URIs, SSL is automatically enabled.
+
+[id="plugins-{type}s-{plugin}-ssl_keystore_path"]
+===== `ssl_keystore_path`
+
+* Value type is <<path,path>>
+* There is no default value for this setting.
+
+The path to the JKS or PKCS12 keystore file for client certificate authentication.
+Use this when the schema registry requires mutual TLS (mTLS) authentication.
+
+[id="plugins-{type}s-{plugin}-ssl_keystore_password"]
+===== `ssl_keystore_password`
+
+* Value type is <<password,password>>
+* There is no default value for this setting.
+
+The password for the keystore file specified in <<plugins-{type}s-{plugin}-ssl_keystore_path>>.
+
+
+[id="plugins-{type}s-{plugin}-ssl_keystore_type"]
+===== `ssl_keystore_type`
+
+* Value type is <<string,string>>
+* There is no default value for this setting.
+
+The format of the keystore file. It must be either `jks` or `pkcs12`.
+
+[id="plugins-{type}s-{plugin}-ssl_supported_protocols"]
+===== `ssl_supported_protocols`
+
+* Value type is <<array,array>>
+* Default value is `[]` (uses Java defaults)
+* Valid values are: `TLSv1.1`, `TLSv1.2`, `TLSv1.3`
+
+List of allowed SSL/TLS protocol versions.
+When not specified, the JVM defaults are used.
+
+[id="plugins-{type}s-{plugin}-ssl_truststore_path"]
+===== `ssl_truststore_path`
+
+* Value type is <<path,path>>
+* There is no default value for this setting.
+
+The path to the JKS or PKCS12 truststore file containing certificates to verify
+the schema registry server's certificate.
+
+*Example*
+[source,ruby]
+----------------------------------
+input {
+  kafka {
+    codec => avro {
+        schema_uri => "https://schema-registry.example.com:8081/schemas/ids/1"
+        ssl_truststore_path => "/path/to/truststore.jks"
+        ssl_truststore_password => "${TRUSTSTORE_PASSWORD}"
+    }
+  }
+}
+----------------------------------
+
+[id="plugins-{type}s-{plugin}-ssl_truststore_password"]
+===== `ssl_truststore_password`
+
+* Value type is <<password,password>>
+* There is no default value for this setting.
+
+The password for the truststore file specified in <<plugins-{type}s-{plugin}-ssl_truststore_path>>.
+
+[id="plugins-{type}s-{plugin}-ssl_truststore_type"]
+===== `ssl_truststore_type`
+
+* Value type is <<string,string>>
+* There is no default value for this setting.
+
+The format of the truststore file. It must be either `jks` or `pkcs12`.
+
+[id="plugins-{type}s-{plugin}-ssl_verification_mode"]
+===== `ssl_verification_mode`
+
+* Value type is <<string,string>>
+* Default value is `"full"`
+* Valid options are: `full`, `none`
+
+Options to verify the server's certificate:
+
+* `full`: Validates that the provided certificate has an issue date that's within the not_before and not_after dates; chains to a trusted Certificate Authority (CA); has a hostname or IP address that matches the names within the certificate. (recommended)
+* `none`: Performs no certificate validation. **Warning:** Disabling this severely compromises security (https://www.cs.utexas.edu/~shmat/shmat_ccs12.pdf)
+
+*Example*
+[source,ruby]
+----------------------------------
+input {
+  kafka {
+    codec => avro {
+        schema_uri => "https://schema-registry.example.com:8081/schemas/ids/1"
+        ssl_certificate_authorities => ["/path/to/ca.crt"]
+        ssl_verification_mode => "full"
+    }
+  }
+}
+----------------------------------
+
 [id="plugins-{type}s-{plugin}-target"]
 ===== `target`
 
@@ -156,3 +355,26 @@ input {
   }
 }
 ----------------------------------
+
+[id="plugins-{type}s-{plugin}-username"]
+===== `username`
+
+* Value type is <<string,string>>
+* There is no default value for this setting.
+
+Username for HTTP basic authentication when fetching remote schemas.
+Used together with `password`.
+
+*Example*
+[source,ruby]
+----------------------------------
+input {
+  kafka {
+    codec => avro {
+        schema_uri => "https://schema-registry.example.com:8081/schemas/ids/1"
+        username => "registry_user"
+        password => "${REGISTRY_PASSWORD}"
+    }
+  }
+}
+----------------------------------