You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@nifi.apache.org by sz...@apache.org on 2024/11/27 17:29:49 UTC
(nifi-minifi-cpp) 03/04: MINIFICPP-2470 Add SSL and mTLS authentication support to CouchbaseClusterService
This is an automated email from the ASF dual-hosted git repository.
szaszm pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/nifi-minifi-cpp.git
commit 3885493eb784a571a8f3f1e06dbf5aca7b221e2a
Author: Gabor Gyimesi <ga...@gmail.com>
AuthorDate: Wed Nov 27 13:05:21 2024 +0100
MINIFICPP-2470 Add SSL and mTLS authentication support to CouchbaseClusterService
- Add mTLS authentication option with linked SSLContextService of CouchbaseClusterService
- Add support for adding CA certificate with linked SSLContextService to use SSL communication when using username/password authentication
- Fix enabling linked services of a controller service before enabling the controller service itself
Closes #1882
Signed-off-by: Marton Szasz <sz...@apache.org>
---
.../cluster/DockerTestDirectoryBindings.py | 10 ++-
.../cluster/containers/CouchbaseServerContainer.py | 88 +++++++++++++++++++---
docker/test/integration/features/couchbase.feature | 57 ++++++++++++++
docker/test/integration/features/steps/steps.py | 30 +++++++-
.../minifi/controllers/CouchbaseClusterService.py | 9 ++-
.../integration/minifi/core/ControllerService.py | 1 +
.../Minifi_flow_json_serializer.py | 13 ++--
.../Minifi_flow_yaml_serializer.py | 38 +++++-----
examples/couchbase_mtls_authentication.json | 74 ++++++++++++++++++
examples/couchbase_mtls_authentication.yml | 45 +++++++++++
.../controllerservices/CouchbaseClusterService.cpp | 66 ++++++++++++++--
.../controllerservices/CouchbaseClusterService.h | 12 +--
.../controller/StandardControllerServiceNode.cpp | 15 +++-
13 files changed, 407 insertions(+), 51 deletions(-)
diff --git a/docker/test/integration/cluster/DockerTestDirectoryBindings.py b/docker/test/integration/cluster/DockerTestDirectoryBindings.py
index 87cf2b4ec..25add90ae 100644
--- a/docker/test/integration/cluster/DockerTestDirectoryBindings.py
+++ b/docker/test/integration/cluster/DockerTestDirectoryBindings.py
@@ -18,7 +18,7 @@ import shutil
import hashlib
import subprocess
import OpenSSL.crypto
-from ssl_utils.SSL_cert_utils import make_self_signed_cert, make_cert_without_extended_usage, make_server_cert
+from ssl_utils.SSL_cert_utils import make_self_signed_cert, make_cert_without_extended_usage, make_server_cert, make_client_cert
class DockerTestDirectoryBindings:
@@ -214,3 +214,11 @@ class DockerTestDirectoryBindings:
os.path.join(base, "root_ca.crt"),
]
subprocess.run(cmd, check=True)
+
+ clientuser_cert, clientuser_key = make_client_cert("clientuser", ca_cert=self.root_ca_cert, ca_key=self.root_ca_key)
+ self.put_test_resource('clientuser.crt',
+ OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM,
+ cert=clientuser_cert))
+ self.put_test_resource('clientuser.key',
+ OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM,
+ pkey=clientuser_key))
diff --git a/docker/test/integration/cluster/containers/CouchbaseServerContainer.py b/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
index 9baf01f08..b2f13b732 100644
--- a/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
+++ b/docker/test/integration/cluster/containers/CouchbaseServerContainer.py
@@ -12,19 +12,67 @@
# 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.
+import os
+import OpenSSL.crypto
+import tempfile
+import docker
+import requests
+import logging
+from requests.auth import HTTPBasicAuth
from .Container import Container
from utils import retry_check
+from ssl_utils.SSL_cert_utils import make_server_cert
class CouchbaseServerContainer(Container):
def __init__(self, feature_context, name, vols, network, image_store, command=None):
- super().__init__(feature_context, name, 'couchbase-server', vols, network, image_store, command)
+ super().__init__(feature_context, name, "couchbase-server", vols, network, image_store, command)
+ couchbase_cert, couchbase_key = make_server_cert(f"couchbase-server-{feature_context.id}", feature_context.root_ca_cert, feature_context.root_ca_key)
+
+ self.root_ca_file = tempfile.NamedTemporaryFile(delete=False)
+ self.root_ca_file.write(OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM, cert=feature_context.root_ca_cert))
+ self.root_ca_file.close()
+ os.chmod(self.root_ca_file.name, 0o666)
+
+ self.couchbase_cert_file = tempfile.NamedTemporaryFile(delete=False)
+ self.couchbase_cert_file.write(OpenSSL.crypto.dump_certificate(type=OpenSSL.crypto.FILETYPE_PEM, cert=couchbase_cert))
+ self.couchbase_cert_file.close()
+ os.chmod(self.couchbase_cert_file.name, 0o666)
+
+ self.couchbase_key_file = tempfile.NamedTemporaryFile(delete=False)
+ self.couchbase_key_file.write(OpenSSL.crypto.dump_privatekey(type=OpenSSL.crypto.FILETYPE_PEM, pkey=couchbase_key))
+ self.couchbase_key_file.close()
+ os.chmod(self.couchbase_key_file.name, 0o666)
def get_startup_finished_log_entry(self):
# after startup the logs are only available in the container, only this message is shown
return "logs available in"
- @retry_check(15, 2)
+ @retry_check(max_tries=12, retry_interval=5)
+ def _run_couchbase_cli_command(self, command):
+ (code, _) = self.client.containers.get(self.name).exec_run(command)
+ if code != 0:
+ logging.error(f"Failed to run command '{command}', returned error code: {code}")
+ return False
+ return True
+
+ def _run_couchbase_cli_commands(self, commands):
+ return all(self._run_couchbase_cli_command(command) for command in commands)
+
+ @retry_check(max_tries=15, retry_interval=2)
+ def _load_couchbase_certs(self):
+ response = requests.post("http://localhost:8091/node/controller/loadTrustedCAs", auth=HTTPBasicAuth("Administrator", "password123"))
+ if response.status_code != 200:
+ logging.error(f"Failed to load CA certificates, with status code: {response.status_code}")
+ return False
+
+ response = requests.post("http://localhost:8091/node/controller/reloadCertificate", auth=HTTPBasicAuth("Administrator", "password123"))
+ if response.status_code != 200:
+ logging.error(f"Failed to reload certificates, with status code: {response.status_code}")
+ return False
+
+ return True
+
def run_post_startup_commands(self):
if self.post_startup_commands_finished:
return True
@@ -33,12 +81,18 @@ class CouchbaseServerContainer(Container):
["couchbase-cli", "cluster-init", "-c", "localhost", "--cluster-username", "Administrator", "--cluster-password", "password123", "--services", "data,index,query",
"--cluster-ramsize", "2048", "--cluster-index-ramsize", "256"],
["couchbase-cli", "bucket-create", "-c", "localhost", "--username", "Administrator", "--password", "password123", "--bucket", "test_bucket", "--bucket-type", "couchbase",
- "--bucket-ramsize", "1024", "--max-ttl", "36000"]
+ "--bucket-ramsize", "1024", "--max-ttl", "36000"],
+ ["couchbase-cli", "user-manage", "-c", "localhost", "-u", "Administrator", "-p", "password123", "--set", "--rbac-username", "clientuser", "--rbac-password", "password123",
+ "--roles", "data_reader[test_bucket],data_writer[test_bucket]", "--auth-domain", "local"],
+ ["bash", "-c", 'tee /tmp/auth.json <<< \'{"state": "enable", "prefixes": [ {"path": "subject.cn", "prefix": "", "delimiter": "."}]}\''],
+ ['couchbase-cli', 'ssl-manage', '-c', 'localhost', '-u', 'Administrator', '-p', 'password123', '--set-client-auth', '/tmp/auth.json']
]
- for command in commands:
- (code, _) = self.client.containers.get(self.name).exec_run(command)
- if code != 0:
- return False
+ if not self._run_couchbase_cli_commands(commands):
+ return False
+
+ if not self._load_couchbase_certs():
+ return False
+
self.post_startup_commands_finished = True
return True
@@ -46,10 +100,26 @@ class CouchbaseServerContainer(Container):
if not self.set_deployed():
return
+ mounts = [
+ docker.types.Mount(
+ type='bind',
+ source=self.couchbase_key_file.name,
+ target='/opt/couchbase/var/lib/couchbase/inbox/pkey.key'),
+ docker.types.Mount(
+ type='bind',
+ source=self.couchbase_cert_file.name,
+ target='/opt/couchbase/var/lib/couchbase/inbox/chain.pem'),
+ docker.types.Mount(
+ type='bind',
+ source=self.root_ca_file.name,
+ target='/opt/couchbase/var/lib/couchbase/inbox/CA/root_ca.crt')
+ ]
+
self.docker_container = self.client.containers.run(
"couchbase:enterprise-7.2.5",
detach=True,
name=self.name,
network=self.network.name,
- ports={'11210/tcp': 11210},
- entrypoint=self.command)
+ ports={'8091/tcp': 8091, '11210/tcp': 11210},
+ entrypoint=self.command,
+ mounts=mounts)
diff --git a/docker/test/integration/features/couchbase.feature b/docker/test/integration/features/couchbase.feature
index 646eaacd9..7b74a6c3d 100644
--- a/docker/test/integration/features/couchbase.feature
+++ b/docker/test/integration/features/couchbase.feature
@@ -173,3 +173,60 @@ Feature: Executing Couchbase operations from MiNiFi-C++
And all instances start up
Then the Minifi logs contain the following message: "Failed to get content for document 'test_doc_id' from collection 'test_bucket._default._default' with the following exception: 'raw_binary_transcoder expects document to have BINARY common flags" in less than 60 seconds
+
+ Scenario: A MiNiFi instance can get data from test bucket with GetCouchbaseKey processor using SSL connection
+ Given a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And the "Keep Source File" property of the GetFile processor is set to "true"
+ And the scheduling period of the GetFile processor is set to "20 seconds"
+ And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
+ And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
+ And the "Couchbase Cluster Controller Service" property of the PutCouchbaseKey processor is set to "CouchbaseClusterService"
+ And a GetCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the GetCouchbaseKey processor is set to "test_doc_id"
+ And the "Couchbase Cluster Controller Service" property of the GetCouchbaseKey processor is set to "CouchbaseClusterService"
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
+ And a CouchbaseClusterService is set up up with SSL connection with the name "CouchbaseClusterService"
+
+ And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
+ And the "success" relationship of the PutCouchbaseKey processor is connected to the GetCouchbaseKey
+ And the "success" relationship of the GetCouchbaseKey processor is connected to the PutFile
+ And the "success" relationship of the PutFile processor is connected to the LogAttribute
+
+ When a Couchbase server is started
+ And all instances start up
+
+ Then a flowfile with the JSON content '{"field1": "value1", "field2": "value2"}' is placed in the monitored directory in less than 6000 seconds
+ And the Minifi logs contain the following message: "key:couchbase.bucket value:test_bucket" in less than 10 seconds
+ And the Minifi logs contain the following message: "key:couchbase.doc.id value:test_doc_id" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.cas value:[1-9][0-9]*" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.expiry value:\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}" in less than 1 seconds
+
+ Scenario: A MiNiFi instance can get data from test bucket with GetCouchbaseKey processor using mTLS authentication
+ Given a MiNiFi CPP server with yaml config
+ And a GetFile processor with the "Input Directory" property set to "/tmp/input"
+ And a file with the content '{"field1": "value1", "field2": "value2"}' is present in '/tmp/input'
+ And a PutCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the PutCouchbaseKey processor is set to "test_doc_id"
+ And the "Couchbase Cluster Controller Service" property of the PutCouchbaseKey processor is set to "CouchbaseClusterService"
+ And a GetCouchbaseKey processor with the "Bucket Name" property set to "test_bucket"
+ And the "Document Id" property of the GetCouchbaseKey processor is set to "test_doc_id"
+ And the "Couchbase Cluster Controller Service" property of the GetCouchbaseKey processor is set to "CouchbaseClusterService"
+ And a PutFile processor with the "Directory" property set to "/tmp/output"
+ And a LogAttribute processor with the "FlowFiles To Log" property set to "0"
+ And a CouchbaseClusterService is setup up using mTLS authentication with the name "CouchbaseClusterService"
+
+ And the "success" relationship of the GetFile processor is connected to the PutCouchbaseKey
+ And the "success" relationship of the PutCouchbaseKey processor is connected to the GetCouchbaseKey
+ And the "success" relationship of the GetCouchbaseKey processor is connected to the PutFile
+ And the "success" relationship of the PutFile processor is connected to the LogAttribute
+
+ When a Couchbase server is started
+ And all instances start up
+
+ Then a flowfile with the JSON content '{"field1": "value1", "field2": "value2"}' is placed in the monitored directory in less than 6000 seconds
+ And the Minifi logs contain the following message: "key:couchbase.bucket value:test_bucket" in less than 10 seconds
+ And the Minifi logs contain the following message: "key:couchbase.doc.id value:test_doc_id" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.cas value:[1-9][0-9]*" in less than 1 seconds
+ And the Minifi logs match the following regex: "key:couchbase.doc.expiry value:\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2}\.\d{3}" in less than 1 seconds
diff --git a/docker/test/integration/features/steps/steps.py b/docker/test/integration/features/steps/steps.py
index 6d4e464e9..70a193888 100644
--- a/docker/test/integration/features/steps/steps.py
+++ b/docker/test/integration/features/steps/steps.py
@@ -573,7 +573,7 @@ def step_impl(context):
minifi_crt_file = '/tmp/resources/minifi_client.crt'
minifi_key_file = '/tmp/resources/minifi_client.key'
root_ca_crt_file = '/tmp/resources/root_ca.crt'
- ssl_context_service = SSLContextService(cert=minifi_crt_file, ca_cert=root_ca_crt_file, key=minifi_key_file)
+ ssl_context_service = SSLContextService(name='SSLContextService', cert=minifi_crt_file, ca_cert=root_ca_crt_file, key=minifi_key_file)
splunk_cert, splunk_key = make_server_cert(context.test.get_container_name_with_postfix("splunk"), context.root_ca_cert, context.root_ca_key)
put_splunk_http = context.test.get_node_by_name("PutSplunkHTTP")
@@ -1384,6 +1384,34 @@ def step_impl(context, service_name):
container.add_controller(couchbase_cluster_controller_service)
+@given("a CouchbaseClusterService is set up up with SSL connection with the name \"{service_name}\"")
+def step_impl(context, service_name):
+ ssl_context_service = SSLContextService(name="SSLContextService",
+ ca_cert='/tmp/resources/root_ca.crt')
+ container = context.test.acquire_container(context=context, name="minifi-cpp-flow")
+ container.add_controller(ssl_context_service)
+ couchbase_cluster_controller_service = CouchbaseClusterService(
+ name=service_name,
+ connection_string="couchbases://{server_hostname}".format(server_hostname=context.test.get_container_name_with_postfix("couchbase-server")),
+ ssl_context_service=ssl_context_service)
+ container.add_controller(couchbase_cluster_controller_service)
+
+
@then("a document with id \"{doc_id}\" in bucket \"{bucket_name}\" is present with data '{data}' of type \"{data_type}\" in Couchbase")
def step_impl(context, doc_id: str, bucket_name: str, data: str, data_type: str):
context.test.check_is_data_present_on_couchbase(doc_id, bucket_name, data, data_type)
+
+
+@given("a CouchbaseClusterService is setup up using mTLS authentication with the name \"{service_name}\"")
+def step_impl(context, service_name):
+ ssl_context_service = SSLContextService(name="SSLContextService",
+ cert='/tmp/resources/clientuser.crt',
+ key='/tmp/resources/clientuser.key',
+ ca_cert='/tmp/resources/root_ca.crt')
+ container = context.test.acquire_container(context=context, name="minifi-cpp-flow")
+ container.add_controller(ssl_context_service)
+ couchbase_cluster_controller_service = CouchbaseClusterService(
+ name=service_name,
+ connection_string="couchbases://{server_hostname}".format(server_hostname=context.test.get_container_name_with_postfix("couchbase-server")),
+ ssl_context_service=ssl_context_service)
+ container.add_controller(couchbase_cluster_controller_service)
diff --git a/docker/test/integration/minifi/controllers/CouchbaseClusterService.py b/docker/test/integration/minifi/controllers/CouchbaseClusterService.py
index e06c1e6ff..94494fe17 100644
--- a/docker/test/integration/minifi/controllers/CouchbaseClusterService.py
+++ b/docker/test/integration/minifi/controllers/CouchbaseClusterService.py
@@ -18,10 +18,13 @@ from ..core.ControllerService import ControllerService
class CouchbaseClusterService(ControllerService):
- def __init__(self, name, connection_string):
+ def __init__(self, name, connection_string, ssl_context_service=None):
super(CouchbaseClusterService, self).__init__(name=name)
self.service_class = 'CouchbaseClusterService'
self.properties['Connection String'] = connection_string
- self.properties['User Name'] = "Administrator"
- self.properties['User Password'] = "password123"
+ if ssl_context_service:
+ self.linked_services.append(ssl_context_service)
+ if not ssl_context_service or ssl_context_service and 'Client Certificate' not in ssl_context_service.properties:
+ self.properties['User Name'] = "Administrator"
+ self.properties['User Password'] = "password123"
diff --git a/docker/test/integration/minifi/core/ControllerService.py b/docker/test/integration/minifi/core/ControllerService.py
index f29513139..02d32e4a7 100644
--- a/docker/test/integration/minifi/core/ControllerService.py
+++ b/docker/test/integration/minifi/core/ControllerService.py
@@ -34,3 +34,4 @@ class ControllerService(object):
properties = {}
self.properties = properties
+ self.linked_services = []
diff --git a/docker/test/integration/minifi/flow_serialization/Minifi_flow_json_serializer.py b/docker/test/integration/minifi/flow_serialization/Minifi_flow_json_serializer.py
index c1043d506..a100ae038 100644
--- a/docker/test/integration/minifi/flow_serialization/Minifi_flow_json_serializer.py
+++ b/docker/test/integration/minifi/flow_serialization/Minifi_flow_json_serializer.py
@@ -118,12 +118,7 @@ class Minifi_flow_json_serializer:
continue
visited.append(svc)
- root['controllerServices'].append({
- 'name': svc.name,
- 'identifier': svc.id,
- 'type': svc.service_class,
- 'properties': svc.properties
- })
+ self.serialize_controller(svc, root)
if isinstance(connectable, Funnel):
root['funnels'].append({
@@ -159,3 +154,9 @@ class Minifi_flow_json_serializer:
'type': controller.service_class,
'properties': controller.properties
})
+
+ if controller.linked_services:
+ if len(controller.linked_services) == 1:
+ root['controllerServices'][-1]['properties']['Linked Services'] = controller.linked_services[0].name
+ else:
+ root['controllerServices'][-1]['properties']['Linked Services'] = [{"value": service.name} for service in controller.linked_services]
diff --git a/docker/test/integration/minifi/flow_serialization/Minifi_flow_yaml_serializer.py b/docker/test/integration/minifi/flow_serialization/Minifi_flow_yaml_serializer.py
index dc912b122..14cbabc98 100644
--- a/docker/test/integration/minifi/flow_serialization/Minifi_flow_yaml_serializer.py
+++ b/docker/test/integration/minifi/flow_serialization/Minifi_flow_yaml_serializer.py
@@ -119,12 +119,7 @@ class Minifi_flow_yaml_serializer:
continue
visited.append(svc)
- res['Controller Services'].append({
- 'name': svc.name,
- 'id': svc.id,
- 'class': svc.service_class,
- 'Properties': svc.properties
- })
+ self._add_controller_service_node(svc, res)
if isinstance(connectable, Funnel):
res['Funnels'].append({
@@ -160,6 +155,25 @@ class Minifi_flow_yaml_serializer:
return (res, visited)
+ def _add_controller_service_node(self, controller, parent):
+ if hasattr(controller, 'name'):
+ connectable_name = controller.name
+ else:
+ connectable_name = str(controller.uuid)
+
+ parent['Controller Services'].append({
+ 'name': connectable_name,
+ 'id': controller.id,
+ 'class': controller.service_class,
+ 'Properties': controller.properties
+ })
+
+ if controller.linked_services:
+ if len(controller.linked_services) == 1:
+ parent['Controller Services'][-1]['Properties']['Linked Services'] = controller.linked_services[0].name
+ else:
+ parent['Controller Services'][-1]['Properties']['Linked Services'] = [{"value": service.name} for service in controller.linked_services]
+
def serialize_controller(self, controller, root=None):
if root is None:
res = {
@@ -175,16 +189,6 @@ class Minifi_flow_yaml_serializer:
else:
res = root
- if hasattr(controller, 'name'):
- connectable_name = controller.name
- else:
- connectable_name = str(controller.uuid)
-
- res['Controller Services'].append({
- 'name': connectable_name,
- 'id': controller.id,
- 'class': controller.service_class,
- 'Properties': controller.properties
- })
+ self._add_controller_service_node(controller, res)
return res
diff --git a/examples/couchbase_mtls_authentication.json b/examples/couchbase_mtls_authentication.json
new file mode 100644
index 000000000..7ac14df25
--- /dev/null
+++ b/examples/couchbase_mtls_authentication.json
@@ -0,0 +1,74 @@
+{
+ "parameterContexts": [],
+ "rootGroup": {
+ "name": "MiNiFi Flow",
+ "processors": [
+ {
+ "name": "Get Couchbase document file from local directory",
+ "identifier": "21b1e56e-e8d5-4543-9f6b-be148f91fb02",
+ "type": "org.apache.nifi.processors.standard.GetFile",
+ "schedulingStrategy": "TIMER_DRIVEN",
+ "schedulingPeriod": "2 sec",
+ "penaltyDuration": "30 sec",
+ "properties": {
+ "Input Directory": "/home/user/couchbase/input"
+ },
+ "autoTerminatedRelationships": [],
+ "concurrentlySchedulableTaskCount": 1
+ },
+ {
+ "name": "Insert Couchbase document",
+ "identifier": "df762d53-0f94-4611-be01-e689b8992573",
+ "type": "org.apache.nifi.processors.standard.PutCouchbaseKey",
+ "schedulingStrategy": "EVENT_DRIVEN",
+ "penaltyDuration": "30 sec",
+ "properties": {
+ "Bucket Name": "test_bucket",
+ "Document Id": "test_doc_id",
+ "Couchbase Cluster Controller Service": "CouchbaseClusterService for mTLS authentication"
+ },
+ "autoTerminatedRelationships": [
+ "success",
+ "failure",
+ "retry"
+ ]
+ }
+ ],
+ "controllerServices": [
+ {
+ "name": "SSLContextService for Couchbase",
+ "identifier": "33e03d54-9917-494e-8ba0-8caeb3fdf4de",
+ "type": "SSLContextService",
+ "properties": {
+ "Client Certificate": "/home/user/couchbase/certs/clientuser.crt",
+ "Private Key": "/home/user/couchbase/certs/clientuser.key",
+ "CA Certificate": "/home/user/couchbase/certsroot_ca.crt"
+ }
+ },
+ {
+ "name": "CouchbaseClusterService for mTLS authentication",
+ "identifier": "747bae3c-e68e-40af-8933-02179bd6cf85",
+ "type": "CouchbaseClusterService",
+ "properties": {
+ "Connection String": "couchbases://couchbase-server-hLkYUYq55djwrW5A26XNJD",
+ "Linked Services": "SSLContextService for Couchbase"
+ }
+ }
+ ],
+ "connections": [
+ {
+ "identifier": "94fdd7b1-7857-44c3-8cf2-d373a5578420",
+ "name": "GetFile/success/PutCouchbaseKey",
+ "source": {
+ "id": "21b1e56e-e8d5-4543-9f6b-be148f91fb02"
+ },
+ "destination": {
+ "id": "df762d53-0f94-4611-be01-e689b8992573"
+ },
+ "selectedRelationships": [
+ "success"
+ ]
+ }
+ ],
+ }
+}
diff --git a/examples/couchbase_mtls_authentication.yml b/examples/couchbase_mtls_authentication.yml
new file mode 100644
index 000000000..7ef1033e7
--- /dev/null
+++ b/examples/couchbase_mtls_authentication.yml
@@ -0,0 +1,45 @@
+Flow Controller:
+ name: MiNiFi Flow
+Processors:
+- id: 21b1e56e-e8d5-4543-9f6b-be148f91fb02
+ name: Get Couchbase document file from local directory
+ class: org.apache.nifi.processors.standard.GetFile
+ scheduling strategy: TIMER_DRIVEN
+ scheduling period: 2 sec
+ penalization period: 30 sec
+ Properties:
+ Input Directory: /tmp/input
+ auto-terminated relationships list: []
+- id: df762d53-0f94-4611-be01-e689b8992573
+ name: Insert Couchbase document
+ class: org.apache.nifi.processors.standard.PutCouchbaseKey
+ scheduling strategy: EVENT_DRIVEN
+ penalization period: 30 sec
+ Properties:
+ Bucket Name: test_bucket
+ Couchbase Cluster Controller Service: CouchbaseClusterService for mTLS authentication
+ Document Id: test_doc_id
+ auto-terminated relationships list:
+ - success
+ - failure
+ - retry
+Controller Services:
+- id: 33e03d54-9917-494e-8ba0-8caeb3fdf4de
+ name: SSLContextService for Couchbase
+ class: SSLContextService
+ Properties:
+ CA Certificate: /tmp/resources/root_ca.crt
+ Client Certificate: /tmp/resources/clientuser.crt
+ Private Key: /tmp/resources/clientuser.key
+- id: 747bae3c-e68e-40af-8933-02179bd6cf85
+ name: CouchbaseClusterService for mTLS authentication
+ class: CouchbaseClusterService
+ Properties:
+ Connection String: couchbases://couchbase-server-VPQDsPD2pj35q5WzHNt9ER
+ Linked Services: SSLContextService for Couchbase
+Connections:
+- id: 94fdd7b1-7857-44c3-8cf2-d373a5578420
+ destination id: df762d53-0f94-4611-be01-e689b8992573
+ name: GetFile/success/PutCouchbaseKey
+ source id: 21b1e56e-e8d5-4543-9f6b-be148f91fb02
+ source relationship name: success
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
index c40697cf7..ac340f864 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.cpp
@@ -50,6 +50,50 @@ CouchbaseErrorType getErrorType(const std::error_code& error_code) {
} // namespace
+CouchbaseClient::CouchbaseClient(std::string connection_string, std::string username, std::string password, minifi::controllers::SSLContextService* ssl_context_service,
+ const std::shared_ptr<core::logging::Logger>& logger)
+ : connection_string_(std::move(connection_string)), logger_(logger), cluster_options_(buildClusterOptions(std::move(username), std::move(password), ssl_context_service)) {
+}
+
+::couchbase::cluster_options CouchbaseClient::buildClusterOptions(std::string username, std::string password, minifi::controllers::SSLContextService* ssl_context_service) {
+ if (username.empty() && (!ssl_context_service || (ssl_context_service && ssl_context_service->getCertificateFile().empty()))) {
+ throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Neither username and password nor SSLContextService is provided for Couchbase authentication");
+ }
+
+ if (!username.empty() && ssl_context_service && !ssl_context_service->getCertificateFile().empty()) {
+ throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Either username and password or mTLS certificate authentication should be used in the SSLContextService for Couchbase, "
+ "but not both");
+ }
+
+ if (!username.empty()) {
+ logger_->log_debug("Using username and password authentication for Couchbase server");
+ if (password.empty()) {
+ throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Password missing for Couchbase server authentication");
+ }
+ ::couchbase::cluster_options cluster_options(std::move(username), std::move(password));
+ if (ssl_context_service && !ssl_context_service->getCACertificate().empty()) {
+ logger_->log_debug("Setting Couchbase client CA certificate path to '{}'", ssl_context_service->getCACertificate().string());
+ cluster_options.security().trust_certificate(ssl_context_service->getCACertificate().string());
+ }
+ return cluster_options;
+ }
+
+ logger_->log_debug("Using mTLS authentication for Couchbase server");
+ logger_->log_debug("Setting Couchbase client SSL key file path to '{}'", ssl_context_service->getPrivateKeyFile().string());
+ logger_->log_debug("Setting Couchbase client certificate file path to '{}'", ssl_context_service->getCertificateFile().string());
+ if (ssl_context_service->getPrivateKeyFile().empty() || ssl_context_service->getCertificateFile().empty()) {
+ throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Couchbase client private key path or client certificate path is empty");
+ }
+
+ ::couchbase::cluster_options cluster_options(::couchbase::certificate_authenticator(ssl_context_service->getCertificateFile().string(), ssl_context_service->getPrivateKeyFile().string()));
+ if (!ssl_context_service->getCACertificate().empty()) {
+ logger_->log_debug("Setting Couchbase client CA certificate path to '{}'", ssl_context_service->getCACertificate().string());
+ cluster_options.security().trust_certificate(ssl_context_service->getCACertificate().string());
+ }
+ cluster_options.security().tls_verify(::couchbase::tls_verify_mode::peer);
+ return cluster_options;
+}
+
nonstd::expected<::couchbase::collection, CouchbaseErrorType> CouchbaseClient::getCollection(const CouchbaseCollection& collection) {
auto connection_result = establishConnection();
if (!connection_result) {
@@ -155,8 +199,7 @@ nonstd::expected<void, CouchbaseErrorType> CouchbaseClient::establishConnection(
return {};
}
- auto options = ::couchbase::cluster_options(username_, password_);
- auto [connect_err, cluster] = ::couchbase::cluster::connect(connection_string_, options).get();
+ auto [connect_err, cluster] = ::couchbase::cluster::connect(connection_string_, cluster_options_).get();
if (connect_err.ec()) {
logger_->log_error("Failed to connect to Couchbase cluster with error code: '{}' and message: '{}'", connect_err.ec(), connect_err.message());
return nonstd::make_unexpected(getErrorType(connect_err.ec()));
@@ -179,11 +222,24 @@ void CouchbaseClusterService::onEnable() {
getProperty(UserName, username);
std::string password;
getProperty(UserPassword, password);
- if (connection_string.empty() || username.empty() || password.empty()) {
- throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Missing connection string, username or password");
+ if (connection_string.empty()) {
+ throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Missing connection string");
+ }
+
+ if ((username.empty() || password.empty()) && linked_services_.empty()) {
+ throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Missing username and password or SSLContextService as a linked service");
+ }
+
+ minifi::controllers::SSLContextService* ssl_context_service_ptr = nullptr;
+ if (!linked_services_.empty()) {
+ auto ssl_context_service = std::dynamic_pointer_cast<minifi::controllers::SSLContextService>(linked_services_[0]);
+ if (!ssl_context_service) {
+ throw minifi::Exception(ExceptionType::PROCESS_SCHEDULE_EXCEPTION, "Linked service is not an SSLContextService");
+ }
+ ssl_context_service_ptr = ssl_context_service.get();
}
+ client_ = std::make_unique<CouchbaseClient>(connection_string, username, password, ssl_context_service_ptr, logger_);
- client_ = std::make_unique<CouchbaseClient>(connection_string, username, password, logger_);
auto result = client_->establishConnection();
if (!result) {
if (result.error() == CouchbaseErrorType::FATAL) {
diff --git a/extensions/couchbase/controllerservices/CouchbaseClusterService.h b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
index 28c6390dd..217449723 100644
--- a/extensions/couchbase/controllerservices/CouchbaseClusterService.h
+++ b/extensions/couchbase/controllerservices/CouchbaseClusterService.h
@@ -31,6 +31,7 @@
#include "couchbase/cluster.hxx"
#include "core/ProcessContext.h"
#include "core/logging/LoggerConfiguration.h"
+#include "controllers/SSLContextService.h"
namespace org::apache::nifi::minifi::couchbase {
@@ -69,9 +70,8 @@ enum class CouchbaseErrorType {
class CouchbaseClient {
public:
- CouchbaseClient(std::string connection_string, std::string username, std::string password, const std::shared_ptr<core::logging::Logger>& logger)
- : connection_string_(std::move(connection_string)), username_(std::move(username)), password_(std::move(password)), logger_(logger) {
- }
+ CouchbaseClient(std::string connection_string, std::string username, std::string password, controllers::SSLContextService* ssl_context_service,
+ const std::shared_ptr<core::logging::Logger>& logger);
~CouchbaseClient() {
close();
@@ -89,14 +89,14 @@ class CouchbaseClient {
void close();
private:
+ ::couchbase::cluster_options buildClusterOptions(std::string username, std::string password, minifi::controllers::SSLContextService* ssl_context_service);
nonstd::expected<::couchbase::collection, CouchbaseErrorType> getCollection(const CouchbaseCollection& collection);
std::string connection_string_;
- std::string username_;
- std::string password_;
+ std::shared_ptr<core::logging::Logger> logger_;
+ ::couchbase::cluster_options cluster_options_;
std::mutex cluster_mutex_;
std::optional<::couchbase::cluster> cluster_;
- std::shared_ptr<core::logging::Logger> logger_;
};
namespace controllers {
diff --git a/libminifi/src/core/controller/StandardControllerServiceNode.cpp b/libminifi/src/core/controller/StandardControllerServiceNode.cpp
index 6bc423b15..0d63420d0 100644
--- a/libminifi/src/core/controller/StandardControllerServiceNode.cpp
+++ b/libminifi/src/core/controller/StandardControllerServiceNode.cpp
@@ -23,11 +23,13 @@
namespace org::apache::nifi::minifi::core::controller {
bool StandardControllerServiceNode::enable() {
- Property property("Linked Services", "Referenced Controller Services");
- controller_service_->setState(ENABLED);
logger_->log_trace("Enabling CSN {}", getName());
+ if (active) {
+ logger_->log_debug("CSN {} is already enabled", getName());
+ return true;
+ }
+ Property property("Linked Services", "Referenced Controller Services");
if (getProperty(property.getName(), property)) {
- active = true;
for (const auto& linked_service : property.getValues()) {
ControllerServiceNode* csNode = provider->getControllerServiceNode(linked_service, controller_service_->getUUID());
if (nullptr != csNode) {
@@ -40,13 +42,20 @@ bool StandardControllerServiceNode::enable() {
if (nullptr != impl) {
std::lock_guard<std::mutex> lock(mutex_);
std::vector<std::shared_ptr<ControllerService>> services;
+ std::vector<ControllerServiceNode*> service_nodes;
services.reserve(linked_controller_services_.size());
for (const auto& service : linked_controller_services_) {
services.push_back(service->getControllerServiceImplementation());
+ if (!service->enable()) {
+ logger_->log_debug("Linked Service '{}' could not be enabled", service->getName());
+ return false;
+ }
}
impl->setLinkedControllerServices(services);
impl->onEnable();
}
+ active = true;
+ controller_service_->setState(ENABLED);
return true;
}