fix(ingestion): robust SSL handling and test fixes

Author: RinZ27

ingestion/src/metadata/ingestion/source/database/sas/client.py

@@ -17,13 +17,17 @@
 from metadata.generated.schema.entity.services.connections.database.sasConnection import (
     SASConnection,
 )
+from metadata.generated.schema.security.ssl.verifySSLConfig import VerifySSL
 from metadata.ingestion.connections.source_api_client import TrackedREST
 from metadata.ingestion.ometa.client import APIError, ClientConfig
 from metadata.utils.helpers import clean_uri
 from metadata.utils.logger import ingestion_logger
+from metadata.utils.ssl_registry import get_verify_ssl_fn
 
 logger = ingestion_logger()
 
+SAS_CLI_AUTH_HEADER = "Basic c2FzLmNsaTo="
+
 
 class SASClient:
     """
@@ -33,13 +37,14 @@ class SASClient:
     def __init__(self, config: SASConnection):
         self.config: SASConnection = config
         self.auth_token = self.get_token(config.serverHost, config.username, config.password.get_secret_value())
+
         client_config: ClientConfig = ClientConfig(
             base_url=clean_uri(config.serverHost),
             auth_header="Authorization",
             auth_token=self.get_auth_token,
             api_version="",
             allow_redirects=True,
-            verify=False,
+            verify=self._get_verify(),
         )
         self.client = TrackedREST(client_config, source_name="sas")
         # custom setting
@@ -50,6 +55,22 @@ def __init__(self, config: SASConnection):
         self.enable_dataflows = config.dataflows
         self.custom_filter_dataflows = config.dataflowsCustomFilter
 
+    def _get_verify(self):
+        """
+        Helper to determine the SSL verification strategy
+        """
+        verify = True
+        if self.config.verifySSL == VerifySSL.ignore:
+            verify = False
+        elif self.config.verifySSL == VerifySSL.no_ssl:
+            verify = False
+        elif self.config.verifySSL == VerifySSL.validate and self.config.sslConfig:
+            try:
+                verify = get_verify_ssl_fn(self.config.verifySSL)(self.config.sslConfig)
+            except Exception:  # pylint: disable=broad-except
+                verify = True
+        return verify
+
     def check_connection(self):
         """
         Check metadata connection to SAS
@@ -71,8 +92,8 @@ def get_instance(self, instance_id):
             "Accept": "application/vnd.sas.metadata.instance.entity.detail+json",
         }
         response = self.client.get(path=endpoint, headers=headers)
-        if "error" in response.keys():  # noqa: SIM118
-            raise APIError(response["error"])
+        if response and isinstance(response, dict) and "error" in response:
+            raise APIError({"message": response["error"]})
         return response
 
     def get_information_catalog_link(self, instance_id):
@@ -98,8 +119,8 @@ def list_assets(self, assets):
         endpoint = f"catalog/search?indices={assets}&q={asset_filter if str(asset_filter) != 'None' else '*'}"
         headers = {"Accept-Item": "application/vnd.sas.metadata.instance.entity+json"}
         response = self.client.get(path=endpoint, headers=headers)
-        if "error" in response.keys():  # noqa: SIM118
-            raise APIError(response["error"])
+        if response and isinstance(response, dict) and "error" in response:
+            raise APIError({"message": response["error"]})
         return response["items"]
 
     def get_views(self, query):
@@ -111,7 +132,7 @@ def get_views(self, query):
         logger.info(f"{query}")
         response = self.client.post(path=endpoint, data=query, headers=headers)
         if "error" in response.keys():  # noqa: SIM118
-            raise APIError(f"{response}")
+            raise APIError({"message": "Error fetching views from SAS"})
         return response
 
     def get_data_source(self, endpoint):
@@ -120,8 +141,8 @@ def get_data_source(self, endpoint):
         }
         response = self.client.get(path=endpoint, headers=headers)
         logger.info(f"{response}")
-        if "error" in response.keys():  # noqa: SIM118
-            raise APIError(response["error"])
+        if response and isinstance(response, dict) and "error" in response:
+            raise APIError({"message": response["error"]})
         return response
 
     def get_report_link(self, resource, uri):
@@ -135,8 +156,8 @@ def load_table(self, endpoint):
     def get_report_relationship(self, report_id):
         endpoint = f"reports/commons/relationships/reports/{report_id}"
         response = self.client.get(endpoint)
-        if "error" in response.keys():  # noqa: SIM118
-            raise APIError(response["error"])
+        if response and isinstance(response, dict) and "error" in response:
+            raise APIError({"message": response["error"]})
         dependencies = []
         for item in response["items"]:
             if item["type"] == "Dependent":
@@ -145,15 +166,15 @@ def get_report_relationship(self, report_id):
 
     def get_resource(self, endpoint):
         response = self.client.get(endpoint)
-        if "error" in response.keys():  # noqa: SIM118
-            raise APIError(response["error"])
+        if response and isinstance(response, dict) and "error" in response:
+            raise APIError({"message": response["error"]})
         return response
 
     def get_instances_with_param(self, data):
         endpoint = f"catalog/instances?{data}"
         response = self.client.get(endpoint)
-        if "error" in response.keys():  # noqa: SIM118
-            raise APIError(response["error"])
+        if response and isinstance(response, dict) and "error" in response:
+            raise APIError({"message": response["error"]})
         return response["items"]
 
     def get_auth_token(self):
@@ -164,8 +185,31 @@ def get_token(self, base_url, user, password):
         payload = {"grant_type": "password", "username": user, "password": password}
         headers = {
             "Content-type": "application/x-www-form-urlencoded",
-            "Authorization": "Basic c2FzLmNsaTo=",
+            "Authorization": SAS_CLI_AUTH_HEADER,
         }
         url = base_url + endpoint
-        response = requests.request("POST", url, headers=headers, data=payload, verify=False, timeout=10)
-        return response.json()["access_token"]
+
+        response = requests.request(
+            "POST",
+            url,
+            headers=headers,
+            data=payload,
+            verify=self._get_verify(),
+            timeout=10,
+        )
+        logger.debug(
+            "Token request for user: %s completed with status: %s",
+            user,
+            response.status_code,
+        )
+        try:
+            body = response.json()
+        except ValueError as exc:
+            response.raise_for_status()
+            raise RuntimeError(f"SAS token endpoint returned non-JSON response (HTTP {response.status_code})") from exc
+
+        response.raise_for_status()
+        token = body.get("access_token")
+        if not token:
+            raise RuntimeError(f"Failed to retrieve access_token from SAS (HTTP {response.status_code})")
+        return token

ingestion/src/metadata/ingestion/source/search/elasticsearch/connection.py

@@ -44,6 +44,7 @@
 from metadata.generated.schema.entity.services.connections.testConnectionResult import (
     TestConnectionResult,
 )
+from metadata.generated.schema.security.ssl.verifySSLConfig import VerifySSL
 from metadata.ingestion.connections.builders import init_empty_connection_arguments
 from metadata.ingestion.connections.test_connections import test_connection_steps
 from metadata.ingestion.ometa.ometa_api import OpenMetadata
@@ -102,22 +103,28 @@ def _handle_ssl_context_by_path(ssl_config: SslConfig):
     return ca_cert, client_cert, private_key
 
 
-def get_ssl_context(ssl_config: SslConfig) -> ssl.SSLContext:
+def get_ssl_context(
+    ssl_config: Optional[SslConfig], verify_ssl: Optional[VerifySSL] = VerifySSL.validate
+) -> Optional[ssl.SSLContext]:
     """
     Method to get SSL Context
     """
+    if verify_ssl == VerifySSL.ignore:
+        return ssl._create_unverified_context()  # pylint: disable=protected-access
+
+    if verify_ssl == VerifySSL.no_ssl:
+        return None
+
     ca_cert = False
     client_cert = None
     private_key = None
     cert_chain = None
 
-    if not ssl_config.certificates:
-        return None
-
-    if isinstance(ssl_config.certificates, SslCertificatesByValues):
-        ca_cert, client_cert, private_key = _handle_ssl_context_by_value(ssl_config=ssl_config)
-    elif isinstance(ssl_config.certificates, SslCertificatesByPath):
-        ca_cert, client_cert, private_key = _handle_ssl_context_by_path(ssl_config=ssl_config)
+    if ssl_config and ssl_config.certificates:
+        if isinstance(ssl_config.certificates, SslCertificatesByValues):
+            ca_cert, client_cert, private_key = _handle_ssl_context_by_value(ssl_config=ssl_config)
+        elif isinstance(ssl_config.certificates, SslCertificatesByPath):
+            ca_cert, client_cert, private_key = _handle_ssl_context_by_path(ssl_config=ssl_config)
 
     if client_cert and private_key:
         cert_chain = (client_cert, private_key)
@@ -133,7 +140,7 @@ def get_ssl_context(ssl_config: SslConfig) -> ssl.SSLContext:
         )
         return ssl_context  # noqa: RET504
 
-    return ssl._create_unverified_context()  # pylint: disable=protected-access
+    return ssl.create_default_context()
 
 
 def get_connection(connection: ElasticsearchConnection) -> Elasticsearch:
@@ -146,7 +153,7 @@ def get_connection(connection: ElasticsearchConnection) -> Elasticsearch:
     if isinstance(connection.authType, BasicAuthentication) and connection.authType.username:
         basic_auth = (
             connection.authType.username,
-            connection.authType.password.get_secret_value() if connection.authType.password else None,
+            (connection.authType.password.get_secret_value() if connection.authType.password else None),
         )
 
     if isinstance(connection.authType, ApiKeyAuthentication):
@@ -161,8 +168,7 @@ def get_connection(connection: ElasticsearchConnection) -> Elasticsearch:
     if not connection.connectionArguments:
         connection.connectionArguments = init_empty_connection_arguments()
 
-    if connection.sslConfig:
-        ssl_context = get_ssl_context(connection.sslConfig)
+    ssl_context = get_ssl_context(connection.sslConfig, connection.verifySSL)
 
     return Elasticsearch(
         str(connection.hostPort),
@@ -188,15 +194,16 @@ def test_connection(
     def test_get_search_indexes():
         try:
             result = client.indices.get_alias(expand_wildcards="open")
-            if result is None:
-                raise ConnectionError("Failed to retrieve search indexes from Elasticsearch")  # noqa: TRY301
-            return result  # noqa: TRY300
         except Exception as exc:
             raise ConnectionError(
                 f"Unable to connect to Elasticsearch or retrieve indexes: {exc}. "
                 "Please check your Elasticsearch connection configuration and cluster health."
             ) from exc
 
+        if result is None:
+            raise ConnectionError("Failed to retrieve search indexes from Elasticsearch")
+        return result
+
     test_fn = {
         "CheckAccess": client.info,
         "GetSearchIndexes": test_get_search_indexes,

ingestion/src/metadata/utils/secrets/aws_secrets_manager.py

@@ -52,7 +52,7 @@ def get_string_value(self, secret_id: str) -> Optional[str]:  # noqa: UP045
         try:
             kwargs = {"SecretId": secret_id}
             response = self.client.get_secret_value(**kwargs)
-            logger.debug("Got value for secret %s.", secret_id)
+            logger.debug("Successfully retrieved value from secrets manager.")
         except ClientError as err:
             logger.debug(traceback.format_exc())
             logger.error(f"Couldn't get value for secret [{secret_id}]: {err}")

ingestion/tests/unit/topology/database/test_sas_client.py

@@ -0,0 +1,125 @@
+#  Copyright 2025 Collate
+#  Licensed under the Collate Community License, Version 1.0 (the "License");
+#  you may not use this file except in compliance with the License.
+#  You may obtain a copy of the License at
+#  https://github.com/open-metadata/OpenMetadata/blob/main/ingestion/LICENSE
+#  Unless required by applicable law or agreed to in writing, software
+#  distributed under the License is distributed on an "AS IS" BASIS,
+#  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.
+"""
+Unit tests for SASClient
+"""
+
+from unittest.mock import MagicMock, patch
+
+import pytest
+from pydantic import SecretStr
+
+from metadata.generated.schema.entity.services.connections.database.sasConnection import (
+    SASConnection,
+)
+from metadata.generated.schema.security.ssl.verifySSLConfig import VerifySSL
+from metadata.ingestion.ometa.client import APIError
+from metadata.ingestion.source.database.sas.client import SASClient
+
+MOCK_CONFIG = SASConnection(
+    username="user",
+    password=SecretStr("pass"),
+    serverHost="http://sas.com",
+    verifySSL=VerifySSL.validate,
+)
+
+
+class TestSASClient:
+    @patch("metadata.ingestion.source.database.sas.client.requests.request")
+    @patch("metadata.ingestion.source.database.sas.client.TrackedREST")
+    def test_init_success(self, mock_rest, mock_request):
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {"access_token": "token123"}
+        mock_request.return_value = mock_response
+
+        client = SASClient(MOCK_CONFIG)
+
+        assert client.auth_token == "token123"
+        mock_rest.assert_called_once()
+        # Verify default verify is True (VerifySSL.validate)
+        args, _ = mock_rest.call_args
+        assert args[0].verify is True
+
+    @patch("metadata.ingestion.source.database.sas.client.requests.request")
+    @patch("metadata.ingestion.source.database.sas.client.TrackedREST")
+    def test_init_verify_ignore(self, mock_rest, mock_request):
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {"access_token": "token123"}
+        mock_request.return_value = mock_response
+
+        config = MOCK_CONFIG.model_copy()
+        config.verifySSL = VerifySSL.ignore
+
+        SASClient(config)
+
+        args, _ = mock_rest.call_args
+        assert args[0].verify is False
+
+    @patch("metadata.ingestion.source.database.sas.client.requests.request")
+    @patch("metadata.ingestion.source.database.sas.client.TrackedREST")
+    def test_get_token_error_handling(self, mock_rest, mock_request):
+        # Case 1: Non-JSON response (with successful HTTP status)
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.raise_for_status.return_value = None
+        mock_response.json.side_effect = ValueError("Not JSON")
+        mock_request.return_value = mock_response
+
+        with pytest.raises(RuntimeError) as exc:
+            SASClient(MOCK_CONFIG)
+        assert "non-JSON response" in str(exc.value)
+
+        # Case 2: Missing access_token
+        mock_response.status_code = 200
+        mock_response.json.side_effect = None
+        mock_response.json.return_value = {"something": "else"}
+        mock_response.raise_for_status.return_value = None
+
+        with pytest.raises(RuntimeError) as exc:
+            SASClient(MOCK_CONFIG)
+        assert "Failed to retrieve access_token" in str(exc.value)
+
+    @patch("metadata.ingestion.source.database.sas.client.requests.request")
+    @patch("metadata.ingestion.source.database.sas.client.TrackedREST")
+    def test_api_error_wrapping(self, mock_rest, mock_request):
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {"access_token": "token123"}
+        mock_request.return_value = mock_response
+
+        client = SASClient(MOCK_CONFIG)
+
+        # Mocking client.get to return an error key
+        client.client.get.return_value = {"error": "Unauthorized access"}
+
+        with pytest.raises(APIError) as exc:
+            client.get_instance("123")
+        assert exc.value.args[0] == "Unauthorized access"
+
+    @patch("metadata.ingestion.source.database.sas.client.requests.request")
+    @patch("metadata.ingestion.source.database.sas.client.TrackedREST")
+    def test_get_views_error_sanitization(self, mock_rest, mock_request):
+        mock_response = MagicMock()
+        mock_response.status_code = 200
+        mock_response.json.return_value = {"access_token": "token123"}
+        mock_request.return_value = mock_response
+
+        client = SASClient(MOCK_CONFIG)
+
+        # Mocking client.post to return an error key
+        client.client.post.return_value = {"error": "Sensitive internal details"}
+
+        with pytest.raises(APIError) as exc:
+            client.get_views("query")
+        # Should show the sanitized message instead of the raw error
+        assert exc.value.args[0] == "Error fetching views from SAS"