Feature: Cassandra SSL

ingestion/src/metadata/examples/workflows/cassandra.yaml

@@ -13,6 +13,11 @@ source:
         #   token: <Token String>
         #   requestTimeout: <Timeout in seconds>
         #   connectTimeout: <Timeout in seconds>
+      # sslMode: allow
+      # sslConfig:
+      #   caCertificate: "CA certificate content"
+      #   sslCertificate: "SSL certificate content"
+      #   sslKey: "SSL key content"
       hostPort: localhost:9042
   sourceConfig:
     config:

ingestion/src/metadata/ingestion/source/database/cassandra/connection.py

@@ -12,7 +12,10 @@
 """
 Source connection handler
 """
+import os
 from functools import partial
+from ssl import CERT_REQUIRED, PROTOCOL_TLS, SSLContext
+from tempfile import NamedTemporaryFile
 from typing import Optional
 
 from cassandra.auth import PlainTextAuthProvider
@@ -34,6 +37,7 @@
 from metadata.generated.schema.entity.services.connections.testConnectionResult import (
     TestConnectionResult,
 )
+from metadata.generated.schema.security.ssl.verifySSLConfig import SslMode
 from metadata.ingestion.connections.test_connections import test_connection_steps
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
 from metadata.ingestion.source.database.cassandra.queries import (
@@ -77,7 +81,35 @@ def get_connection(connection: CassandraConnection):
                 password=connection.authType.password.get_secret_value(),
             )
 
-    cluster = Cluster(**cluster_config)
+    ssl_context = None
+    if connection.sslMode != SslMode.disable:
+        ssl_context = SSLContext(PROTOCOL_TLS)
+
+        # Load CA certificate directly into memory
+        ssl_context.load_verify_locations(
+            cadata=connection.sslConfig.root.caCertificate.get_secret_value()
+        )
+
+        ssl_context.verify_mode = CERT_REQUIRED
+
+        # Create temporary files since the load_cert_chain function requires
+        # file paths for the certfile and keyfile
+        with NamedTemporaryFile(delete=False, mode="w") as certfile, NamedTemporaryFile(
+            delete=False, mode="w"
+        ) as keyfile:
+            certfile.write(connection.sslConfig.root.sslCertificate.get_secret_value())
+            certfile_path = certfile.name
+
+            keyfile.write(connection.sslConfig.root.sslKey.get_secret_value())
+            keyfile_path = keyfile.name
+
+        ssl_context.load_cert_chain(certfile=certfile_path, keyfile=keyfile_path)
+
+        # Delete temporary files
+        os.remove(certfile_path)
+        os.remove(keyfile_path)
+
+    cluster = Cluster(**cluster_config, ssl_context=ssl_context)
     session = cluster.connect()
 
     return session

ingestion/src/metadata/ingestion/source/database/cassandra/helpers.py

@@ -9,7 +9,7 @@
 #  See the License for the specific language governing permissions and
 #  limitations under the License.
 """
-Iceberg source helpers.
+Cassandra source helpers.
 """
 from __future__ import annotations
 

openmetadata-service/src/main/java/org/openmetadata/service/jdbi3/MlModelRepository.java

@@ -211,7 +211,8 @@ private void setMlFeatureSourcesLineage(MlModel mlModel) {
                       .getFeatureSources()
                       .forEach(
                           mlFeatureSource -> {
-                            EntityReference targetEntity = getEntityReference(mlFeatureSource.getDataSource(), Include.ALL);
+                            EntityReference targetEntity =
+                                getEntityReference(mlFeatureSource.getDataSource(), Include.ALL);
                             if (targetEntity != null) {
                               addRelationship(
                                   targetEntity.getId(),

openmetadata-spec/src/main/resources/json/schema/entity/services/connections/database/cassandraConnection.json

@@ -49,6 +49,14 @@
             "description": "Optional name to give to the database in OpenMetadata. If left blank, we will use default as the database name.",
             "type": "string"
         },
+        "sslMode": {
+            "title": "SSL Mode",
+            "$ref": "../../../../security/ssl/verifySSLConfig.json#/definitions/sslMode"
+        },
+        "sslConfig": {
+            "title": "SSL Configuration",
+            "$ref": "../../../../security/ssl/verifySSLConfig.json#/definitions/sslConfig"
+        },
         "supportsMetadataExtraction": {
             "title": "Supports Metadata Extraction",
             "$ref": "../connectionBasicType.json#/definitions/supportsMetadataExtraction"

openmetadata-ui/src/main/resources/ui/public/locales/en-US/Database/Cassandra.md

@@ -29,13 +29,27 @@ $$
 
 $$section
 ### Cloud Config $(id="cloudConfig")
-
-  Configuration for connecting to DataStax Astra DB in the cloud.
+Configuration settings required when connecting to DataStax Astra DB in the cloud environment. These settings help establish and maintain secure connections to your cloud-hosted Cassandra database.
+$$
+
+$$section
+### Connect Timeout $(id="connectTimeout")
+Specifies the timeout duration in seconds for establishing new connections to Cassandra. This setting helps control how long the system should wait when attempting to create a new connection before timing out.
+$$
+
+$$section
+### Request Timeout $(id="requestTimeout")
+Defines the timeout duration in seconds for individual Cassandra requests. This setting determines how long each query or operation should wait for a response before timing out.
+$$
+
+$$section
+### Token $(id="token")
+The authentication token required for connecting to DataStax Astra DB. This token serves as the security credential for accessing your cloud database instance.
+$$
 
-  - connectTimeout: Timeout in seconds for establishing new connections to Cassandra.
-  - requestTimeout: Timeout in seconds for individual Cassandra requests.
-  - token: The Astra DB application token used for authentication.
-  - secureConnectBundle: File path to the Secure Connect Bundle (.zip) used for a secure connection to DataStax Astra DB.
+$$section
+### Secure Connect Bundle $(id="secureConnectBundle")
+The file path to the Secure Connect Bundle (.zip) file. This bundle contains the necessary certificates and configuration files required to establish a secure connection to your DataStax Astra DB instance.
 $$
 
 $$section
@@ -46,3 +60,16 @@ Database Service > Database > Schema > Table
 ```
 In the case of Cassandra, we won't have a Keyspace/Database as such. If you'd like to see your data in a database named something other than `default`, you can specify the name in this field.
 $$
+
+$$section
+### SSL Mode $(id="sslMode")
+SSL Mode to connect to Cassandra instance. By default, SSL is disabled.
+$$
+
+$$section
+### SSL Configuration $(id="sslConfig")
+SSL Configuration for the Cassandra connection. This is required when SSL Mode is enabled.
+- `CA Certificate`: Path to the CA certificate file.
+- `SSL Certificate`: Path to the client certificate file.
+- `SSL Key`: Path to the client private key file.
+$$