diff --git a/ingestion/src/metadata/ingestion/source/database/sas/client.py b/ingestion/src/metadata/ingestion/source/database/sas/client.py
index d7f82f07ca1d..6197bde5daf0 100644
--- a/ingestion/src/metadata/ingestion/source/database/sas/client.py
+++ b/ingestion/src/metadata/ingestion/source/database/sas/client.py
@@ -12,25 +12,21 @@
 Client to interact with SAS Viya apis
 """
 
-import os
-
 import requests
 
 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()
 
-# Default-secure: verify TLS unless the operator explicitly opts out via
-# OM_SAS_VERIFY_SSL=false (e.g. dev deployments with self-signed certs).
-# Replaces the previous hard-coded verify=False, which Snyk Code flags as
-# python/SSLVerificationBypass.
-_VERIFY_SSL = os.environ.get("OM_SAS_VERIFY_SSL", "true").lower() not in ("false", "0", "no")
+SAS_CLI_AUTH_HEADER = "Basic c2FzLmNsaTo="
 
 
 class SASClient:
@@ -40,14 +36,21 @@ 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())
+        verify = self._get_verify()
+        self.auth_token = self.get_token(
+            config.serverHost,
+            config.username,
+            config.password.get_secret_value(),
+            verify=verify,
+        )
+
         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=_VERIFY_SSL,
+            verify=verify,
         )
         self.client = TrackedREST(client_config, source_name="sas")
         # custom setting
@@ -58,6 +61,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
@@ -79,8 +98,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"], "code": 0})
         return response
 
     def get_information_catalog_link(self, instance_id):
@@ -101,13 +120,18 @@ def list_assets(self, assets):
             asset_filter = self.custom_filter_dataflows
 
         logger.debug(
-            f"Configuration for {assets}: enable {assets} - {enable_asset}, custom {assets} filter - {asset_filter}"
+            "Configuration for %s: enable %s - %s, custom %s filter - %s",
+            assets,
+            assets,
+            enable_asset,
+            assets,
+            asset_filter,
         )
         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"], "code": 0})
         return response["items"]
 
     def get_views(self, query):
@@ -116,10 +140,10 @@ def get_views(self, query):
             "Content-type": "application/vnd.sas.metadata.instance.query+json",
             "Accept": "application/json",
         }
-        logger.info(f"{query}")
+        logger.info("%s", query)
         response = self.client.post(path=endpoint, data=query, headers=headers)
-        if "error" in response.keys():  # noqa: SIM118
-            raise APIError(f"{response}")
+        if response and isinstance(response, dict) and "error" in response:
+            raise APIError({"message": "Error fetching views from SAS", "code": 0})
         return response
 
     def get_data_source(self, endpoint):
@@ -127,9 +151,9 @@ def get_data_source(self, endpoint):
             "Accept-Item": "application/vnd.sas.data.source+json",
         }
         response = self.client.get(path=endpoint, headers=headers)
-        logger.info(f"{response}")
-        if "error" in response.keys():  # noqa: SIM118
-            raise APIError(response["error"])
+        logger.info("%s", response)
+        if response and isinstance(response, dict) and "error" in response:
+            raise APIError({"message": response["error"], "code": 0})
         return response
 
     def get_report_link(self, resource, uri):
@@ -143,8 +167,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"], "code": 0})
         dependencies = []
         for item in response["items"]:
             if item["type"] == "Dependent":
@@ -153,27 +177,50 @@ 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"], "code": 0})
         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"], "code": 0})
         return response["items"]
 
     def get_auth_token(self):
         return self.auth_token, 0
 
-    def get_token(self, base_url, user, password):
+    def get_token(self, base_url, user, password, verify=True):
         endpoint = "/SASLogon/oauth/token"
         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=_VERIFY_SSL, timeout=10)
-        return response.json()["access_token"]
+
+        response = requests.request(
+            "POST",
+            url,
+            headers=headers,
+            data=payload,
+            verify=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
diff --git a/ingestion/src/metadata/ingestion/source/search/elasticsearch/connection.py b/ingestion/src/metadata/ingestion/source/search/elasticsearch/connection.py
index 2253314a73b5..ace76f8a768d 100644
--- a/ingestion/src/metadata/ingestion/source/search/elasticsearch/connection.py
+++ b/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: SslConfig | None, verify_ssl: VerifySSL | None = VerifySSL.validate
+) -> ssl.SSLContext | None:
     """
     Method to get SSL Context
     """
+    if verify_ssl == VerifySSL.no_ssl:
+        return None
+
+    if verify_ssl == VerifySSL.ignore:
+        return ssl._create_unverified_context()
+
     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)
@@ -127,13 +134,12 @@ def get_ssl_context(ssl_config: SslConfig) -> ssl.SSLContext:
         cert_chain = None
 
     if ca_cert or cert_chain:
-        ssl_context = create_ssl_context(
+        return create_ssl_context(
             cert=cert_chain,
             verify=ca_cert,
         )
-        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:
@@ -161,8 +167,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),
diff --git a/ingestion/src/metadata/utils/secrets/aws_secrets_manager.py b/ingestion/src/metadata/utils/secrets/aws_secrets_manager.py
index 15552c7bd18f..3976e2afb9a8 100644
--- a/ingestion/src/metadata/utils/secrets/aws_secrets_manager.py
+++ b/ingestion/src/metadata/utils/secrets/aws_secrets_manager.py
@@ -1,4 +1,4 @@
-#  Copyright 2022 Collate
+#  Copyright 2021 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
@@ -10,7 +10,7 @@
 #  limitations under the License.
 
 """
-Secrets manager implementation using AWS Secrets Manager
+AWS Secrets Manager handle
 """
 
 import traceback
@@ -24,27 +24,32 @@
 from metadata.generated.schema.security.secrets.secretsManagerProvider import (
     SecretsManagerProvider,
 )
+from metadata.utils.logger import ingestion_logger
 from metadata.utils.secrets.aws_based_secrets_manager import (
     NULL_VALUE,
     AWSBasedSecretsManager,
 )
-from metadata.utils.secrets.secrets_manager import logger
+
+logger = ingestion_logger()
 
 
 class AWSSecretsManager(AWSBasedSecretsManager):
     """
-    Secrets Manager Implementation Class
+    AWS Secrets Manager
     """
 
     def __init__(self, loader: SecretsManagerClientLoader):
-        super().__init__(client="secretsmanager", provider=SecretsManagerProvider.aws, loader=loader)
+        super().__init__(
+            client="secretsmanager",
+            provider=SecretsManagerProvider.aws,
+            loader=loader,
+        )
 
     def get_string_value(self, secret_id: str) -> Optional[str]:  # noqa: UP045
         """
-        :param secret_id: The secret id to retrieve. Current stage is always retrieved.
-        :return: The value of the secret. When the secret is a string, the value is
-                 contained in the `SecretString` field. When the secret is bytes or not present,
-                 it throws a `ValueError` exception.
+        Get the secret value as a string
+        :param secret_id: secret id
+        :return: secret value
         """
         if secret_id is None:
             raise ValueError("[name] argument is None")
@@ -55,7 +60,7 @@ def get_string_value(self, secret_id: str) -> Optional[str]:  # noqa: UP045
             logger.debug("Got value for secret %s.", secret_id)
         except ClientError as err:
             logger.debug(traceback.format_exc())
-            logger.error(f"Couldn't get value for secret [{secret_id}]: {err}")
+            logger.error(f"Couldn't get value for secret [{secret_id}] from secrets manager: {err}")
             raise err  # noqa: TRY201
         if "SecretString" in response:
             return response["SecretString"] if response["SecretString"] != NULL_VALUE else None
diff --git a/openmetadata-integration-tests/src/test/java/org/openmetadata/it/factories/SearchServiceTestFactory.java b/openmetadata-integration-tests/src/test/java/org/openmetadata/it/factories/SearchServiceTestFactory.java
index d88ed3d129da..fb4eeeef2d3d 100644
--- a/openmetadata-integration-tests/src/test/java/org/openmetadata/it/factories/SearchServiceTestFactory.java
+++ b/openmetadata-integration-tests/src/test/java/org/openmetadata/it/factories/SearchServiceTestFactory.java
@@ -7,6 +7,7 @@
 import org.openmetadata.schema.api.services.CreateSearchService;
 import org.openmetadata.schema.api.services.CreateSearchService.SearchServiceType;
 import org.openmetadata.schema.entity.services.SearchService;
+import org.openmetadata.schema.security.ssl.VerifySSL;
 import org.openmetadata.schema.services.connections.search.ElasticSearchConnection;
 import org.openmetadata.schema.services.connections.search.OpenSearchConnection;
 import org.openmetadata.schema.type.SearchConnection;
@@ -27,7 +28,9 @@ public static SearchService createElasticSearch(TestNamespace ns) {
     String name = ns.prefix("elasticService_" + uniqueId);
 
     ElasticSearchConnection esConn =
-        new ElasticSearchConnection().withHostPort(URI.create("http://localhost:9200"));
+        new ElasticSearchConnection()
+            .withHostPort(URI.create("http://localhost:9200"))
+            .withVerifySSL(VerifySSL.IGNORE);
 
     SearchConnection conn = new SearchConnection().withConfig(esConn);
 
diff --git a/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/SearchServiceResourceIT.java b/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/SearchServiceResourceIT.java
index fb9433577278..6c26bd1aa09a 100644
--- a/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/SearchServiceResourceIT.java
+++ b/openmetadata-integration-tests/src/test/java/org/openmetadata/it/tests/SearchServiceResourceIT.java
@@ -147,7 +147,8 @@ void post_searchServiceWithElasticSearchConnection_200_OK(TestNamespace ns) {
     ElasticSearchConnection conn =
         new ElasticSearchConnection()
             .withHostPort(URI.create("http://localhost:9200"))
-            .withAuthType(auth);
+            .withAuthType(auth)
+            .withVerifySSL(VerifySSL.IGNORE);
 
     CreateSearchService request =
         new CreateSearchService()
diff --git a/openmetadata-spec/src/main/resources/json/schema/entity/services/connections/database/sasConnection.json b/openmetadata-spec/src/main/resources/json/schema/entity/services/connections/database/sasConnection.json
index d256996d8e69..dc1641120c39 100644
--- a/openmetadata-spec/src/main/resources/json/schema/entity/services/connections/database/sasConnection.json
+++ b/openmetadata-spec/src/main/resources/json/schema/entity/services/connections/database/sasConnection.json
@@ -114,6 +114,16 @@
     "supportsMetadataExtraction": {
       "title": "Supports Metadata Extraction",
       "$ref": "../connectionBasicType.json#/definitions/supportsMetadataExtraction"
+    },
+    "sslConfig": {
+      "title": "SSL Config",
+      "$ref": "../../../../security/ssl/verifySSLConfig.json#/definitions/sslConfig"
+    },
+    "verifySSL": {
+      "title": "Verify SSL",
+      "description": "Client SSL verification. Make sure to configure the SSLConfig if enabled.",
+      "$ref": "../../../../security/ssl/verifySSLConfig.json#/definitions/verifySSL",
+      "default": "ignore"
     }
   },
   "required": ["username", "password", "serverHost"],
diff --git a/openmetadata-spec/src/main/resources/json/schema/entity/services/connections/search/elasticSearchConnection.json b/openmetadata-spec/src/main/resources/json/schema/entity/services/connections/search/elasticSearchConnection.json
index ef58725d0887..5922f66a97da 100644
--- a/openmetadata-spec/src/main/resources/json/schema/entity/services/connections/search/elasticSearchConnection.json
+++ b/openmetadata-spec/src/main/resources/json/schema/entity/services/connections/search/elasticSearchConnection.json
@@ -38,6 +38,10 @@
         }
       ]
     },
+    "verifySSL": {
+      "$ref": "../../../../security/ssl/verifySSLConfig.json#/definitions/verifySSL",
+      "default": "ignore"
+    },
     "sslConfig": {
       "title": "SSL Config",
       "$ref": "../common/sslConfig.json"
diff --git a/openmetadata-ui/src/main/resources/ui/src/generated/api/services/createDatabaseService.ts b/openmetadata-ui/src/main/resources/ui/src/generated/api/services/createDatabaseService.ts
index ea9cbf205bd7..51f519391b75 100644
--- a/openmetadata-ui/src/main/resources/ui/src/generated/api/services/createDatabaseService.ts
+++ b/openmetadata-ui/src/main/resources/ui/src/generated/api/services/createDatabaseService.ts
@@ -910,6 +910,8 @@ export interface Connection {
     tokenUrl?: string;
     /**
      * Client SSL verification.
+     *
+     * Client SSL verification. Make sure to configure the SSLConfig if enabled.
      */
     verifySSL?: VerifySSL;
     /**
diff --git a/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/connections/database/sasConnection.ts b/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/connections/database/sasConnection.ts
index 18769912c943..2314161d596b 100644
--- a/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/connections/database/sasConnection.ts
+++ b/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/connections/database/sasConnection.ts
@@ -54,6 +54,7 @@ export interface SASConnection {
      * Hostname of SAS Viya deployment.
      */
     serverHost:                  string;
+    sslConfig?:                  Config;
     supportsMetadataExtraction?: boolean;
     /**
      * Regex to only include/exclude tables that matches the pattern.
@@ -67,6 +68,10 @@ export interface SASConnection {
      * Username to connect to SAS Viya.
      */
     username: string;
+    /**
+     * Client SSL verification. Make sure to configure the SSLConfig if enabled.
+     */
+    verifySSL?: VerifySSL;
 }
 
 /**
@@ -89,6 +94,26 @@ export interface FilterPattern {
     includes?: string[];
 }
 
+/**
+ * Client SSL configuration
+ *
+ * OpenMetadata Client configured to validate SSL certificates.
+ */
+export interface Config {
+    /**
+     * The CA certificate used for SSL validation.
+     */
+    caCertificate?: string;
+    /**
+     * The SSL certificate used for client authentication.
+     */
+    sslCertificate?: string;
+    /**
+     * The private key associated with the SSL certificate.
+     */
+    sslKey?: string;
+}
+
 /**
  * Service Type
  *
@@ -97,3 +122,12 @@ export interface FilterPattern {
 export enum SASType {
     SAS = "SAS",
 }
+
+/**
+ * Client SSL verification. Make sure to configure the SSLConfig if enabled.
+ */
+export enum VerifySSL {
+    Ignore = "ignore",
+    NoSSL = "no-ssl",
+    Validate = "validate",
+}
diff --git a/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/connections/search/elasticSearchConnection.ts b/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/connections/search/elasticSearchConnection.ts
index 26aaea74d745..a96526cf27eb 100644
--- a/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/connections/search/elasticSearchConnection.ts
+++ b/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/connections/search/elasticSearchConnection.ts
@@ -36,7 +36,8 @@ export interface ElasticSearchConnection {
     /**
      * ElasticSearch Type
      */
-    type?: ElasticSearchType;
+    type?:      ElasticSearchType;
+    verifySSL?: VerifySSL;
 }
 
 /**
@@ -138,3 +139,12 @@ export interface SSLCertificates {
 export enum ElasticSearchType {
     ElasticSearch = "ElasticSearch",
 }
+
+/**
+ * Client SSL verification. Make sure to configure the SSLConfig if enabled.
+ */
+export enum VerifySSL {
+    Ignore = "ignore",
+    NoSSL = "no-ssl",
+    Validate = "validate",
+}
diff --git a/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/databaseService.ts b/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/databaseService.ts
index 8acaef365c54..6172b742e529 100644
--- a/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/databaseService.ts
+++ b/openmetadata-ui/src/main/resources/ui/src/generated/entity/services/databaseService.ts
@@ -1041,6 +1041,8 @@ export interface Connection {
     tokenUrl?: string;
     /**
      * Client SSL verification.
+     *
+     * Client SSL verification. Make sure to configure the SSLConfig if enabled.
      */
     verifySSL?: VerifySSL;
     /**