Initial Draft version to load the CA trusted store code.

Signed-off-by: lrangine <19699092+lokeshrangineni@users.noreply.github.com>
feast-dev · franciscojavierarceo · Dec 18, 2024 · Dec 14, 2024 · Dec 14, 2024 · Dec 14, 2024
commit f11ce849edade251b4f3c8615cbf60c4429b00e0
@@ -49,6 +49,7 @@
     teardown,
 )
 from feast.saved_dataset import SavedDataset, ValidationReference
+from feast.ssl_ca_setup import configure_ssl_ca
 from feast.utils import maybe_local_tz
 
 _logger = logging.getLogger(__name__)
@@ -865,7 +866,6 @@ def materialize_incremental_command(ctx: click.Context, end_ts: str, views: List
             "cassandra",
             "hazelcast",
             "ikv",
-            "couchbase",
         ],
         case_sensitive=False,
     ),
@@ -956,6 +956,15 @@ def init_command(project_directory, minimal: bool, template: str):
     show_default=False,
     help="path to TLS certificate public key. You need to pass --key as well to start server in TLS mode",
 )
+@click.option(
+    "--ca",
+    "-ca",
+    "tls_ca_file_path",
+    type=click.STRING,
+    default="",
+    show_default=False,
+    help="path to ca trust store file. This will override the global path set as part of the environment variable FEAST_CA_CERT_FILE_PATH",
+)
 @click.option(
     "--metrics",
     "-m",
@@ -975,6 +984,7 @@ def serve_command(
     keep_alive_timeout: int,
     tls_key_path: str,
     tls_cert_path: str,
+    tls_ca_file_path: str,
     registry_ttl_sec: int = 5,
 ):
     """Start a feature server locally on a given port."""
@@ -983,6 +993,8 @@ def serve_command(
             "Please pass --cert and --key args to start the feature server in TLS mode."
         )
 
+    configure_ssl_ca(ca_file_path=tls_ca_file_path)
+
     store = create_feature_store(ctx)
 
     store.serve(
@@ -1082,14 +1094,25 @@ def serve_transformations_command(ctx: click.Context, port: int):
     show_default=False,
     help="path to TLS certificate public key. You need to pass --key as well to start server in TLS mode",
 )
+@click.option(
+    "--ca",
+    "-ca",
+    "tls_ca_file_path",
+    type=click.STRING,
+    default="",
+    show_default=False,
+    help="path to ca trust store file. This will override the global path set as part of the environment variable FEAST_CA_CERT_FILE_PATH",
+)
 @click.pass_context
 def serve_registry_command(
     ctx: click.Context,
     port: int,
     tls_key_path: str,
     tls_cert_path: str,
+    tls_ca_file_path: str,
 ):
     """Start a registry server locally on a given port."""
+    configure_ssl_ca(ca_file_path=tls_ca_file_path)
     if (tls_key_path and not tls_cert_path) or (not tls_key_path and tls_cert_path):
         raise click.BadParameter(
             "Please pass --cert and --key args to start the registry server in TLS mode."
@@ -1134,13 +1157,13 @@ def serve_registry_command(
     help="path to TLS certificate public key. You need to pass --key as well to start server in TLS mode",
 )
 @click.option(
-    "--verify_client",
-    "-v",
-    "tls_verify_client",
-    type=click.BOOL,
-    default="True",
-    show_default=True,
-    help="Verify the client or not for the TLS client certificate.",
+    "--ca",
+    "-ca",
+    "tls_ca_file_path",
+    type=click.STRING,
+    default="",
+    show_default=False,
+    help="path to ca trust store file. This will override the global path set as part of the environment variable FEAST_CA_CERT_FILE_PATH",
 )
 @click.pass_context
 def serve_offline_command(
@@ -1149,16 +1172,17 @@ def serve_offline_command(
     port: int,
     tls_key_path: str,
     tls_cert_path: str,
-    tls_verify_client: bool,
+    tls_ca_file_path: str,
 ):
     """Start a remote server locally on a given host, port."""
+    configure_ssl_ca(ca_file_path=tls_ca_file_path)
     if (tls_key_path and not tls_cert_path) or (not tls_key_path and tls_cert_path):
         raise click.BadParameter(
             "Please pass --cert and --key args to start the offline server in TLS mode."
         )
     store = create_feature_store(ctx)
 
-    store.serve_offline(host, port, tls_key_path, tls_cert_path, tls_verify_client)
+    store.serve_offline(host, port, tls_key_path, tls_cert_path)
 
 
 @cli.command("validate")

@@ -1,3 +1,5 @@
+import os
+import ssl
 from datetime import datetime
 from pathlib import Path
 from typing import List, Optional, Union
@@ -6,6 +8,7 @@
 from google.protobuf.empty_pb2 import Empty
 from google.protobuf.timestamp_pb2 import Timestamp
 from pydantic import StrictStr
+from sqlglot.expressions import false
 
 from feast.base_feature_view import BaseFeatureView
 from feast.data_source import DataSource
@@ -59,6 +62,11 @@ class RemoteRegistryConfig(RegistryConfig):
     """ str: Path to the public certificate when the registry server starts in TLS(SSL) mode. This may be needed if the registry server started with a self-signed certificate, typically this file ends with `*.crt`, `*.cer`, or `*.pem`.
     If registry_type is 'remote', then this configuration is needed to connect to remote registry server in TLS mode. If the remote registry started in non-tls mode then this configuration is not needed."""
 
+    is_tls_mode: bool = False
+    """ str: Path to the public certificate when the registry server starts in TLS(SSL) mode. This may be needed if the registry server started with a self-signed certificate, typically this file ends with `*.crt`, `*.cer`, or `*.pem`.
+    If registry_type is 'remote', then this configuration is needed to connect to remote registry server in TLS mode. If the remote registry started in non-tls mode then this configuration is not needed."""
+
+
 
 class RemoteRegistry(BaseRegistry):
     def __init__(
@@ -70,20 +78,49 @@ def __init__(
     ):
         self.auth_config = auth_config
         assert isinstance(registry_config, RemoteRegistryConfig)
-        if registry_config.cert:
-            with open(registry_config.cert, "rb") as cert_file:
-                trusted_certs = cert_file.read()
-                tls_credentials = grpc.ssl_channel_credentials(
-                    root_certificates=trusted_certs
-                )
-            self.channel = grpc.secure_channel(registry_config.path, tls_credentials)
-        else:
-            self.channel = grpc.insecure_channel(registry_config.path)
+        # self.channel = create_tls_channel(registry_config)
+
+        self._create_grpc_channel(registry_config)
 
         auth_header_interceptor = GrpcClientAuthHeaderInterceptor(auth_config)
         self.channel = grpc.intercept_channel(self.channel, auth_header_interceptor)
         self.stub = RegistryServer_pb2_grpc.RegistryServerStub(self.channel)
 
+    def _create_grpc_channel(self, registry_config):
+        assert isinstance(registry_config, RemoteRegistryConfig)
+        if registry_config.cert or registry_config.is_tls_mode:
+            cafile = os.getenv("SSL_CERT_FILE") or os.getenv("REQUESTS_CA_BUNDLE")
+            if not cafile and not registry_config.cert:
+                raise EnvironmentError(
+                    "SSL_CERT_FILE or REQUESTS_CA_BUNDLE environment variable must be set to use secure TLS or set the cert parameter in feature_Store.yaml file under remote registry configuration."
+                )
+
+            with open(registry_config.cert if registry_config.cert else cafile, "rb") as cert_file:
+                trusted_certs = cert_file.read()
+            tls_credentials = grpc.ssl_channel_credentials(
+                root_certificates=trusted_certs
+            )
+            self.channel = grpc.secure_channel(registry_config.path, tls_credentials)
+        # elif registry_config.is_tls_mode:
+        #     # Use the trust store path from the environment variable
+        #     cafile = os.getenv("SSL_CERT_FILE") or os.getenv("REQUESTS_CA_BUNDLE")
+        #     if cafile:
+        #         # Load the CA certificates from the trust store
+        #         with open(cafile, "rb") as cert_file:
+        #             root_certificates = cert_file.read()
+        #
+        #             # Create TLS credentials using the root certificates
+        #         tls_credentials = grpc.ssl_channel_credentials(root_certificates=root_certificates)
+        #         # Create a secure gRPC channel
+        #         self.channel = grpc.secure_channel(registry_config.path, tls_credentials)
+        #     else:
+        #         raise EnvironmentError(
+        #             "SSL_CERT_FILE or REQUESTS_CA_BUNDLE environment variable must be set to use secure TLS or set the cert parameter in feature_Store.yaml file under remote registry configuration."
+        #         )
+        else:
+            # Create an insecure gRPC channel
+            self.channel = grpc.insecure_channel(registry_config.path)
+
     def close(self):
         if self.channel:
             self.channel.close()

diff --git a/sdk/python/feast/ssl_ca_setup.py b/sdk/python/feast/ssl_ca_setup.py
@@ -0,0 +1,20 @@
+import logging
+import os
+
+logger = logging.getLogger(__name__)
+logger.setLevel(logging.INFO)
+
+def configure_ssl_ca(ca_file_path:str= ""):
+    """
+       configures the environment variable so that other libraries or servers refer the TLS ca file path.
+       :param ca_file_path:
+       :return:
+       """
+    if ca_file_path:
+      os.environ["SSL_CERT_FILE"] = ca_file_path
+      os.environ["REQUESTS_CA_BUNDLE"] = ca_file_path
+    elif 'FEAST_CA_CERT_FILE_PATH' in os.environ and os.environ['FEAST_CA_CERT_FILE_PATH']:
+      logger.info(f"CA Cert file path found in environment variable FEAST_CA_CERT_FILE_PATH={os.environ['FEAST_CA_CERT_FILE_PATH']}")
+      os.environ["SSL_CERT_FILE"] = os.environ['FEAST_CA_CERT_FILE_PATH']
+      os.environ["REQUESTS_CA_BUNDLE"] = os.environ['FEAST_CA_CERT_FILE_PATH']
+
@@ -29,6 +29,7 @@
 
 from feast.data_source import DataSource
 from feast.feature_store import FeatureStore  # noqa: E402
+from feast.ssl_ca_setup import configure_ssl_ca
 from feast.utils import _utc_now
 from feast.wait import wait_retry_backoff  # noqa: E402
 from tests.data.data_creator import (
@@ -57,8 +58,8 @@
     location,
 )
 from tests.utils.auth_permissions_util import default_store
-from tests.utils.generate_self_signed_certifcate_util import generate_self_signed_cert
 from tests.utils.http_server import check_port_open, free_port  # noqa: E402
+from tests.utils.ssl_certifcates_util import create_ca_trust_store, generate_self_signed_cert
 
 logger = logging.getLogger(__name__)
 
@@ -514,17 +515,24 @@ def auth_config(request, is_integration_test):
     return auth_configuration
 
 
-@pytest.fixture(params=[True, False], scope="module")
+@pytest.fixture(scope="module")
 def tls_mode(request):
-    is_tls_mode = request.param
+    is_tls_mode = request.param[0]
+    ca_trust_store_path = ""
 
     if is_tls_mode:
         certificates_path = tempfile.mkdtemp()
         tls_key_path = os.path.join(certificates_path, "key.pem")
         tls_cert_path = os.path.join(certificates_path, "cert.pem")
+
         generate_self_signed_cert(cert_path=tls_cert_path, key_path=tls_key_path)
+        is_ca_trust_store_set = request.param[1]
+        if is_ca_trust_store_set:
+            ca_trust_store_path = os.path.join(certificates_path, "ca_trust_store.pem")
+            create_ca_trust_store(public_key_path=tls_cert_path, private_key_path=tls_key_path, output_trust_store_path=ca_trust_store_path)
+            configure_ssl_ca(ca_file_path=ca_trust_store_path)
     else:
         tls_key_path = ""
         tls_cert_path = ""
 
-    return is_tls_mode, tls_key_path, tls_cert_path
+    return is_tls_mode, tls_key_path, tls_cert_path, ca_trust_store_path
@@ -34,7 +34,6 @@
     DataSourceCreator,
 )
 from tests.utils.auth_permissions_util import include_auth_config
-from tests.utils.generate_self_signed_certifcate_util import generate_self_signed_cert
 from tests.utils.http_server import check_port_open, free_port  # noqa: E402
 
 logger = logging.getLogger(__name__)

@@ -22,6 +22,11 @@
 
 
 @pytest.mark.integration
+@pytest.mark.parametrize("tls_mode", [
+    ("True", "True"),
+    ("True", "False"),
+    ("False", "")
+], indirect=True)
 def test_remote_online_store_read(auth_config, tls_mode):
     with (
         tempfile.TemporaryDirectory() as remote_server_tmp_dir,
@@ -56,13 +61,13 @@ def test_remote_online_store_read(auth_config, tls_mode):
             )
         )
         assert None not in (server_store, server_url, registry_path)
-        _, _, tls_cert_path = tls_mode
+
         client_store = _create_remote_client_feature_store(
             temp_dir=remote_client_tmp_dir,
             server_registry_path=str(registry_path),
             feature_server_url=server_url,
             auth_config=auth_config,
-            tls_cert_path=tls_cert_path,
+            tls_mode=tls_mode,
         )
         assert client_store is not None
         _assert_non_existing_entity_feature_views_entity(
@@ -172,7 +177,7 @@ def _create_server_store_spin_feature_server(
 ):
     store = default_store(str(temp_dir), auth_config, permissions_list)
     feast_server_port = free_port()
-    is_tls_mode, tls_key_path, tls_cert_path = tls_mode
+    is_tls_mode, tls_key_path, tls_cert_path, _ = tls_mode
 
     server_url = next(
         start_feature_server(
@@ -203,20 +208,29 @@ def _create_remote_client_feature_store(
     server_registry_path: str,
     feature_server_url: str,
     auth_config: str,
-    tls_cert_path: str = "",
+    tls_mode
 ) -> FeatureStore:
     project_name = "REMOTE_ONLINE_CLIENT_PROJECT"
     runner = CliRunner()
     result = runner.run(["init", project_name], cwd=temp_dir)
     assert result.returncode == 0
     repo_path = os.path.join(temp_dir, project_name, "feature_repo")
-    _overwrite_remote_client_feature_store_yaml(
-        repo_path=str(repo_path),
-        registry_path=server_registry_path,
-        feature_server_url=feature_server_url,
-        auth_config=auth_config,
-        tls_cert_path=tls_cert_path,
-    )
+    _, _, tls_cert_path, ca_trust_store_path = tls_mode
+    if ca_trust_store_path:
+        _overwrite_remote_client_feature_store_yaml(
+            repo_path=str(repo_path),
+            registry_path=server_registry_path,
+            feature_server_url=feature_server_url,
+            auth_config=auth_config
+        )
+    else:
+        _overwrite_remote_client_feature_store_yaml(
+            repo_path=str(repo_path),
+            registry_path=server_registry_path,
+            feature_server_url=feature_server_url,
+            auth_config=auth_config,
+            tls_cert_path=tls_cert_path,
+        )
 
     return FeatureStore(repo_path=repo_path)
 

@@ -13,6 +13,7 @@
 )
 from feast.permissions.permission import Permission
 from feast.registry_server import start_server
+from feast.ssl_ca_setup import configure_ssl_ca
 from feast.wait import wait_retry_backoff  # noqa: E402
 from tests.unit.permissions.auth.server import mock_utils
 from tests.unit.permissions.auth.server.test_utils import (
@@ -44,8 +45,10 @@ def start_registry_server(
 
     assertpy.assert_that(server_port).is_not_equal_to(0)
 
-    is_tls_mode, tls_key_path, tls_cert_path = tls_mode
+    is_tls_mode, tls_key_path, tls_cert_path, tls_ca_file_path= tls_mode
     if is_tls_mode:
+        #configure_ssl_ca(ca_file_path=tls_ca_file_path)
+        # Setting the ca_trust_store_path environment variables.
         print(f"Starting Registry in TLS mode at {server_port}")
         server = start_server(
             store=feature_store,
@@ -74,6 +77,11 @@ def start_registry_server(
     server.stop(grace=None)  # Teardown server
 
 
+@pytest.mark.parametrize("tls_mode", [
+    ("True", "True"),
+    ("True", "False"),
+    ("False", "")
+], indirect=True)
 def test_registry_apis(
     auth_config,
     tls_mode,