DOCSP-40133: custom auth provider (#163)

* DOCSP-40133: custom auth provider

* wip

* vale

* fixes

* updates

* add to wn

* wip RW tech comments

* RW comments

* fix

* fix

* RW comments

(cherry picked from commit e21986b)

source/includes/security/AwsAssumeRoleCredentialProvider.java

@@ -0,0 +1,54 @@
+package com.mongodb;
+
+import java.util.Map;
+import java.util.function.Supplier;
+
+import com.mongodb.kafka.connect.util.custom.credentials.CustomCredentialProvider;
+
+import com.amazonaws.auth.AWSCredentialsProvider;
+import com.amazonaws.auth.DefaultAWSCredentialsProviderChain;
+import com.amazonaws.services.securitytoken.AWSSecurityTokenService;
+import com.amazonaws.services.securitytoken.AWSSecurityTokenServiceAsyncClientBuilder;
+import com.amazonaws.services.securitytoken.model.AssumeRoleRequest;
+import com.amazonaws.services.securitytoken.model.AssumeRoleResult;
+import com.amazonaws.services.securitytoken.model.Credentials;
+import com.amazonaws.util.StringUtils;
+
+public class SampleAssumeRoleCredential implements CustomCredentialProvider {
+
+  public SampleAssumeRoleCredential() {}
+  @Override
+  public MongoCredential getCustomCredential(Map<?, ?> map) {
+    AWSCredentialsProvider provider = new DefaultAWSCredentialsProviderChain();
+    Supplier<AwsCredential> awsFreshCredentialSupplier = () -> {
+      AWSSecurityTokenService stsClient = AWSSecurityTokenServiceAsyncClientBuilder.standard()
+          .withCredentials(provider)
+          .withRegion("us-east-1")
+          .build();
+      AssumeRoleRequest assumeRoleRequest = new AssumeRoleRequest().withDurationSeconds(3600)
+          .withRoleArn((String)map.get("mongodbaws.auth.mechanism.roleArn"))
+          .withRoleSessionName("Test_Session");
+      AssumeRoleResult assumeRoleResult = stsClient.assumeRole(assumeRoleRequest);
+      Credentials creds = assumeRoleResult.getCredentials();
+      // Add your code to fetch new credentials
+      return new AwsCredential(creds.getAccessKeyId(), creds.getSecretAccessKey(), creds.getSessionToken());
+    };
+    return MongoCredential.createAwsCredential(null, null)
+        .withMechanismProperty(MongoCredential.AWS_CREDENTIAL_PROVIDER_KEY, awsFreshCredentialSupplier);
+  }
+
+  // Validates presence of an ARN
+  @Override
+  public void validate(Map<?, ?> map) {
+    String roleArn = (String) map.get("mongodbaws.auth.mechanism.roleArn");
+    if (StringUtils.isNullOrEmpty(roleArn)) {
+      throw new RuntimeException("Invalid value set for customProperty");
+    }
+  }
+
+  // Initializes the custom provider
+  @Override
+  public void init(Map<?, ?> map) {
+
+  }
+}

source/security-and-authentication.txt

@@ -10,10 +10,11 @@ Security and Authentication
 
    SSL/TLS and X.509 Certificates </security-and-authentication/tls-and-x509>
    MongoDB AWS-based Authentication </security-and-authentication/mongodb-aws-auth>
+   Custom Authentication Provider </security-and-authentication/custom-auth>
 
 Read the following sections to learn how to secure communications between MongoDB 
 and the {+connector+}:
 
-- :doc:`Encrypt the Messages Your Connector Sends with SSL/TLS </security-and-authentication/tls-and-x509>`
-- :doc:`Authenticate Your Connector with MongoDB using Amazon Web Services </security-and-authentication/mongodb-aws-auth>`
-
+- :ref:`Encrypt the Messages Your Connector Sends with SSL/TLS <kafka-configure-ssl>`
+- :ref:`Authenticate Your Connector with MongoDB using Amazon Web Services <kafka-mongodb-aws>`
+- :ref:`Implement a Custom Authentication Provider <kafka-custom-auth>`

source/security-and-authentication/custom-auth.txt

@@ -0,0 +1,83 @@
+.. _kafka-custom-auth:
+
+==============================
+Custom Authentication Provider
+==============================
+
+.. facet::
+   :name: genre
+   :values: reference
+
+.. meta::
+   :keywords: credentials, implementation class, custom class
+
+.. contents:: On this page
+   :local:
+   :backlinks: none
+   :depth: 2
+   :class: singlecol
+
+Overview
+--------
+
+You can add a custom authentication provider by implementing the
+``com.mongodb.kafka.connect.util.custom.credentials.CustomCredentialProvider``
+interface. You must place your custom class JAR file in the ``lib`` folder
+in your {+kafka-connect+} deployment.
+
+Set following authentication properties to configure the authentication
+provider:
+
+- ``mongo.custom.auth.mechanism.enable``: set to ``true``
+- ``mongo.custom.auth.mechanism.providerClass``: set to the qualified
+  class name of the implementation class
+- *(Optional)* ``mongodbaws.auth.mechanism.roleArn``: set to an Amazon Resource Name (ARN)
+
+AWS IAM Authentication Example
+------------------------------
+
+This example provides a custom authentication provider that supports AWS
+IAM. The following code shows the custom authentication
+provider JAR file:
+
+.. literalinclude:: /includes/security/AwsAssumeRoleCredentialProvider.java
+   :language: java
+
+Compile the JAR file and place it in the ``lib`` folder in your
+deployment. 
+
+.. note::
+
+   To view an example of a ``pom.xml`` file that can build the complete JAR containing
+   the implementation class, see the `Kafka Connector GitHub repository
+   README file
+   <https://github.com/mongodb/mongo-kafka/blob/master/README.md#pom-file-to-build-the-sample-customroleprovider-into-a-jar>`__.
+
+Next, configure your source or sink connector to include the custom
+authentication method. The following configuration properties define a
+sink connector that connects the {+connector-short+} to MongoDB Atlas
+by using AWS IAM authentication:
+
+.. code-block:: json
+   :emphasize-lines: 12-14
+
+   {
+      "name": "mongo-tutorial-sink",
+      "config": {
+        "connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
+        "topics": "<topic>",
+        "connection.uri": "<connection string>?authSource=%24external&authMechanism=MONGODB-AWS&retryWrites=true&w=majority",
+        "key.converter": "org.apache.kafka.connect.storage.StringConverter",
+        "value.converter": "org.apache.kafka.connect.json.JsonConverter",
+        "value.converter.schemas.enable": false,
+        "database": "<db>",
+        "collection": "<collection>",
+        "mongo.custom.auth.mechanism.enable": "true",
+        "mongo.custom.auth.mechanism.providerClass": "com.mongodb.SampleAssumeRoleCredential",
+        "mongodbaws.auth.mechanism.roleArn": "<AWS IAM ARN>"
+      }
+   }
+
+In this example, the ``roleArn`` value is the IAM Role of the user group that has
+access to MongoDB Atlas. In the AWS IAM console, the IAM account that is
+running {+kafka-connect+} has ``AssumeRole`` permissions to the Atlas User Group.

source/security-and-authentication/mongodb-aws-auth.txt

@@ -79,3 +79,8 @@ replace:
          and placeholder value.
        | *Optional*
 
+.. tip:: Custom Authentication Provider
+
+   You can create and use a custom authentication provider to support
+   AWS IAM authentication. To learn more, see the
+   :ref:`kafka-custom-auth` guide.

source/whats-new.txt

@@ -47,9 +47,7 @@ What's New in 1.13
 
 - Added a custom authentication provider interface for Source and Sink
   Connectors. This feature enables you to write and use a custom implementation
-  class in your connector.
-
-.. TODO add link To learn more, see the :ref:`` guide.
+  class in your connector. To learn more, see the :ref:`kafka-custom-auth` guide.
 
 - Fixed an issue that occurred when validating configuration for Source
   and Sink Connectors if the configuration contained secrets and used