Fix KEEPER_SSL_CERT_FILE ignored for HTTP SSL verification

keepercommander/api.py

@@ -754,7 +754,7 @@ def execute_router_rest(params: KeeperParams, endpoint: str, payload: Optional[b
     if payload is not None:
         payload = crypto.encrypt_aes_v2(payload, transmission_key)
     rs = requests.post(url, data=payload, headers=headers, proxies=params.rest_context.proxies,
-                       verify=params.rest_context.certificate_check)
+                       verify=params.ssl_verify)
     if rs.status_code == 200:
         rs_body = rs.content
         if rs_body:

keepercommander/attachment.py

@@ -278,7 +278,7 @@ def upload_attachments(params, record, attachments):
                 }
                 response = requests.post(uo.url, files=files, data=json.loads(uo.parameters),
                                          proxies=params.rest_context.proxies,
-                                         verify=params.rest_context.certificate_check)
+                                         verify=params.ssl_verify)
                 if response.status_code == uo.success_status_code:
                     facade.file_ref.append(file_ref)
                     if record.linked_keys is None:
@@ -295,7 +295,7 @@ def upload_attachments(params, record, attachments):
                         }
                         requests.post(uo.url, files=files, data=json.loads(uo.thumbnail_parameters),
                                       proxies=params.rest_context.proxies,
-                                      verify=params.rest_context.certificate_check)
+                                      verify=params.ssl_verify)
                 except Exception as e:
                     logging.warning('Error uploading thumbnail: %s', e)
     else:

keepercommander/commands/pam/router_helper.py

@@ -19,8 +19,6 @@
 from ...params import KeeperParams
 from ...proto import pam_pb2, router_pb2
 
-VERIFY_SSL = bool(os.environ.get("VERIFY_SSL", "TRUE") == "TRUE")
-
 
 # `RouterResponseError` lives in `_layer_b` to avoid the pre-existing circular
 # import chain (gateway_helper -> commands.utils -> ksm -> record). It's raised
@@ -162,7 +160,7 @@ def _post_request_to_router(params, path, rq_proto=None, rs_type=None, method='p
         rs = requests.request(method,
                               krouter_host + path,
                               params=query_params,
-                              verify=VERIFY_SSL,
+                              verify=params.ssl_verify,
                               headers={
                                 'TransmissionKey': bytes_to_base64(encrypted_transmission_key),
                                 'Authorization': f'KeeperUser {bytes_to_base64(encrypted_session_token)}'
@@ -359,14 +357,14 @@ def router_send_message_to_gateway(params, transmission_key, rq_proto,
     if http_session is not None:
         rs = http_session.post(
             krouter_host + "/api/user/send_controller_message",
-            verify=VERIFY_SSL,
+            verify=params.ssl_verify,
             headers=headers,
             data=encrypted_payload if rq_proto else None
         )
     else:
         rs = requests.post(
             krouter_host + "/api/user/send_controller_message",
-            verify=VERIFY_SSL,
+            verify=params.ssl_verify,
             headers=headers,
             data=encrypted_payload if rq_proto else None
         )
@@ -406,7 +404,7 @@ def get_dag_leafs(params, encrypted_session_token, encrypted_transmission_key, r
     try:
         rs = requests.request('post',
                               krouter_host + path,
-                              verify=VERIFY_SSL,
+                              verify=params.ssl_verify,
                               headers={
                                   'TransmissionKey': bytes_to_base64(encrypted_transmission_key),
                                   'Authorization': f'KeeperUser {bytes_to_base64(encrypted_session_token)}'

keepercommander/commands/pam_debug/krouter.py

@@ -46,16 +46,15 @@ def execute(self, params: KeeperParams, **kwargs):
             url = get_router_url(params)
         url = url.rstrip('/') + '/healthcheck'
 
-        verify_ssl = os.environ.get('VERIFY_SSL', 'TRUE') == 'TRUE'
         timeout = kwargs.get('timeout') or 10.0
         raw = bool(kwargs.get('raw'))
 
         try:
-            rs = requests.get(url, verify=verify_ssl, timeout=timeout)
+            rs = requests.get(url, verify=params.ssl_verify, timeout=timeout)
             rs.raise_for_status()
         except requests.exceptions.SSLError as err:
             print(f"{bcolors.FAIL}SSL verification failed: {err}{bcolors.ENDC}")
-            print('  Set VERIFY_SSL=FALSE to bypass for development krouter builds.')
+            print('  Set VERIFY_SSL=FALSE or KEEPER_SSL_CERT_FILE=none to bypass SSL verification.')
             return
         except requests.exceptions.ConnectionError as err:
             print(f"{bcolors.FAIL}Cannot reach krouter at {url}: {err}{bcolors.ENDC}")

keepercommander/commands/pam_launch/terminal_connection.py

@@ -65,7 +65,6 @@
     router_send_action_to_gateway,
     router_get_relay_access_creds,
     get_router_url,
-    VERIFY_SSL,
 )
 from ...proto import pam_pb2
 from ...display import bcolors
@@ -213,7 +212,7 @@ def _notify_gateway_connection_close(params, router_token, terminated=True):
         response = requests.post(
             f"{router_url}/api/device/connect_state",
             json=payload,
-            verify=VERIFY_SSL,
+            verify=params.ssl_verify,
             timeout=10,
         )
         if response.status_code >= 400:
@@ -1360,7 +1359,7 @@ def _open_terminal_webrtc_tunnel(params: KeeperParams,
             krouter_host = get_router_url(params)
             try:
                 bind_url = krouter_host + "/api/user/bind_to_controller/" + gateway_uid
-                http_session.get(bind_url, verify=VERIFY_SSL, timeout=10)
+                http_session.get(bind_url, verify=params.ssl_verify, timeout=10)
             except Exception as e:
                 logging.debug("bind_to_controller GET failed (continuing): %s", e)
             if http_session.cookies:

keepercommander/commands/recordv3.py

@@ -380,7 +380,7 @@ def GCM_TAG_LEN(): return 16
                     logging.info('Uploading %s ...', file['full_path'])
                     response = requests.post(url, data=form_params, files=form_files,
                                              proxies=params.rest_context.proxies,
-                                             verify=params.rest_context.certificate_check)
+                                             verify=params.ssl_verify)
                     if 'success_action_status' in form_params and str(response.status_code) == form_params['success_action_status']:
                         attachments.append(file)
                         # params.queue_audit_event('file_attachment_uploaded', record_uid=record_uid, attachment_id=a['file_id'])

keepercommander/commands/sso_cloud/metadata_commands.py

@@ -112,7 +112,7 @@ def execute(self, params, **kwargs):
         rs = http_requests.get(
             metadata_url,
             proxies=params.rest_context.proxies,
-            verify=params.rest_context.certificate_check,
+            verify=params.ssl_verify,
             timeout=30,
         )
         if rs.status_code != 200:

keepercommander/commands/tunnel/port_forward/tunnel_helpers.py

@@ -174,6 +174,7 @@ def print_above_keeper_prompt(msg):
 READ_TIMEOUT = 1.5
 KRELAY_URL = 'KRELAY_SERVER'
 GATEWAY_TIMEOUT = int(os.getenv('GATEWAY_TIMEOUT')) if os.getenv('GATEWAY_TIMEOUT') else 30000
+# VERIFY_SSL applies to WebSocket SSL only; HTTP uses params.ssl_verify.
 VERIFY_SSL = bool(os.environ.get("VERIFY_SSL", "TRUE") == "TRUE")
 
 # ICE candidate buffering - store until SDP answer is received
@@ -2475,10 +2476,9 @@ def start_rust_tunnel(params, record_uid, gateway_uid, host, port,
             logging.debug("Using shared router tokens for WebSocket and streaming HTTP")
             http_session = requests.Session()
             krouter_host = get_router_url(params)
-            verify_ssl = bool(os.environ.get("VERIFY_SSL", "TRUE").upper() == "TRUE")
             try:
                 bind_url = krouter_host + "/api/user/bind_to_controller/" + gateway_uid
-                http_session.get(bind_url, verify=verify_ssl, timeout=10)
+                http_session.get(bind_url, verify=params.ssl_verify, timeout=10)
             except Exception as e:
                 logging.debug(f"bind_to_controller GET failed (continuing): %s", e)
             if http_session.cookies:

keepercommander/importer/imp_exp.py

@@ -1411,7 +1411,7 @@ def upload_v3_attachments(params, records_with_attachments):  # type: (KeeperPar
                     print(f'{atta.name} ... ', file=sys.stderr, end='', flush=True)
                     response = requests.post(f.url, data=form_params, files=form_files,
                                              proxies=params.rest_context.proxies,
-                                             verify=params.rest_context.certificate_check)
+                                             verify=params.ssl_verify)
 
             if str(response.status_code) == form_params.get('success_action_status'):
                 print('Done')

keepercommander/importer/lastpass/lastpass.py

@@ -137,7 +137,7 @@ def do_import(self, name, users_only=False, old_domain=None, new_domain=None, tm
         params = kwargs.get('params')
         if isinstance(params, KeeperParams):
             request_settings['proxies'] = params.rest_context.proxies
-            request_settings['certificate_check'] = params.rest_context.certificate_check
+            request_settings['certificate_check'] = params.ssl_verify
         if 'filter_folder' in kwargs and kwargs['filter_folder']:
             request_settings['filter_folder'] = kwargs['filter_folder']
 
@@ -711,7 +711,7 @@ def download_membership(self, params, **kwargs):
         session = None
         request_settings = {
             'proxies': params.rest_context.proxies,
-            'certificate_check': params.rest_context.certificate_check
+            'certificate_check': params.ssl_verify
         }
         try:
             session = fetcher.login(username, password, twofa_code, **request_settings)

keepercommander/keeper_dag/connection/commander.py

@@ -45,7 +45,7 @@ def __init__(self,
                          use_write_protobuf=use_write_protobuf)
 
         self.params = params
-        self.verify_ssl = value_to_boolean(os.environ.get("VERIFY_SSL", verify_ssl))
+        self.verify_ssl = False if verify_ssl is False else utils.resolve_http_ssl_verify(params)
         self.is_ws = is_ws
 
         # Deprecated; setting this will override the per-transaction values.

keepercommander/keeper_dag/connection/ksm.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 from . import ConnectionBase
 from ..utils import value_to_boolean
+from ... import utils as commander_utils
 from ..exceptions import DAGException, DAGConnectionException
 
 from cryptography.hazmat.primitives import hashes
@@ -70,11 +71,9 @@ def __init__(self,
         if isinstance(config, dict) is False and isinstance(config, KeyValueStorage) is False:
             raise DAGException("The configuration is not a dictionary.")
 
-        if verify_ssl is None:
-            verify_ssl = value_to_boolean(os.environ.get("VERIFY_SSL", "TRUE"))
+        self.verify_ssl = False if verify_ssl is False else commander_utils.resolve_http_ssl_verify()
 
         self.config = config
-        self.verify_ssl = verify_ssl
         self._signature = None
         self._challenge_str = None
 

keepercommander/params.py

@@ -45,6 +45,7 @@ def __init__(self, server='https://keepersecurity.com/api/v2/', locale='en_US'):
         self.__store_server_key = False
         self.proxies = None
         self._certificate_check = True
+        self._resolved_verify = None
         self.fail_on_throttle = False
         self.client_ec_private_key = None  # EC private key for QRC ECDH exchange
 
@@ -133,12 +134,17 @@ def set_proxy(self, proxy_server):
 
     @property
     def certificate_check(self):
-        return self._certificate_check
+        if self._resolved_verify is None:
+            from . import utils
+            self._resolved_verify = utils.resolve_ssl_verify(
+                certificate_check_enabled=self._certificate_check)
+        return self._resolved_verify
 
     @certificate_check.setter
     def certificate_check(self, value):
         if isinstance(value, bool):
             self._certificate_check = value
+            self._resolved_verify = None
             if value:
                 warnings.simplefilter('default', InsecureRequestWarning)
             else:
@@ -363,6 +369,10 @@ def queue_audit_event(self, name, **kwargs):
     server = property(__get_server, __set_server)
     rest_context = property(__get_rest_context)
 
+    @property
+    def ssl_verify(self):
+        return self.rest_context.certificate_check
+
     def is_feature_disallowed(self, feature_name):    # type: (str) -> bool
         return isinstance(self.disallowed_features, list) and feature_name in self.disallowed_features
 

keepercommander/utils.py

@@ -530,31 +530,42 @@ def value_to_boolean(value):
     else:
         return None
 
+_SSL_CERT_UNSET = object()
+_cached_ssl_cert_file = _SSL_CERT_UNSET
+
+
+def _legacy_ssl_verify_disabled():
+    return os.environ.get('VERIFY_SSL', 'TRUE').upper() == 'FALSE'
+
+
 def get_ssl_cert_file():
-    """Resolve the SSL CA bundle path.
+    """Resolve KEEPER_SSL_CERT_FILE to a CA bundle path, or False to disable verification."""
+    global _cached_ssl_cert_file
+    if _cached_ssl_cert_file is not _SSL_CERT_UNSET:
+        return _cached_ssl_cert_file
 
-    KEEPER_SSL_CERT_FILE accepts: 'certifi', 'system', 'none'/'false', or a PEM path.
-    Defaults to the bundled certifi store; does not consult SSL_CERT_FILE from the
-    environment to avoid silent trust-store hijacking by unrelated tools.
-    """
     import certifi
 
     user_cert_file = os.getenv('KEEPER_SSL_CERT_FILE')
     if user_cert_file:
         choice = user_cert_file.lower()
         if choice == 'certifi':
-            return certifi.where()
+            _cached_ssl_cert_file = certifi.where()
+            return _cached_ssl_cert_file
         if choice in ('none', 'false'):
-            return False
+            _cached_ssl_cert_file = False
+            return _cached_ssl_cert_file
         if choice != 'system':
             if os.path.exists(user_cert_file):
-                return user_cert_file
+                _cached_ssl_cert_file = user_cert_file
+                return _cached_ssl_cert_file
             print(
                 f"Warning: KEEPER_SSL_CERT_FILE points to a non-existent file: "
                 f"{user_cert_file}; falling back to the bundled certifi store.",
                 file=sys.stderr,
             )
-            return certifi.where()
+            _cached_ssl_cert_file = certifi.where()
+            return _cached_ssl_cert_file
 
     try:
         system = platform.system()
@@ -575,33 +586,46 @@ def get_ssl_cert_file():
             candidates = ()
         for ca_path in candidates:
             if os.path.exists(ca_path):
-                return ca_path
+                _cached_ssl_cert_file = ca_path
+                return _cached_ssl_cert_file
     except Exception:
         pass
 
-    return certifi.where()
+    _cached_ssl_cert_file = certifi.where()
+    return _cached_ssl_cert_file
+
+
+def resolve_ssl_verify(*, certificate_check_enabled=True):
+    """Return the ``requests`` ``verify`` value (False or a CA bundle path)."""
+    if _legacy_ssl_verify_disabled():
+        return False
+    if not certificate_check_enabled:
+        return False
+    return get_ssl_cert_file()
+
+
+def resolve_http_ssl_verify(params=None):
+    """Resolve HTTP SSL verification for Commander or standalone DAG clients."""
+    if params is not None:
+        return params.ssl_verify
+    return resolve_ssl_verify()
 
 
 def ssl_aware_request(method, url, **kwargs):
     """Make an SSL-aware HTTP request using system CA certificates when available"""
     import requests
-
-    # Only set verify if not already specified
+
     if 'verify' not in kwargs:
-        cert_file = get_ssl_cert_file()
-        if cert_file is False:
-            kwargs['verify'] = False
-        elif cert_file:
-            kwargs['verify'] = cert_file
-        # If cert_file is None, let requests use its default
-
+        kwargs['verify'] = resolve_ssl_verify()
+
     return requests.request(method, url, **kwargs)
 
 
 def ssl_aware_get(url, **kwargs):
     """SSL-aware GET request using system CA certificates when available"""
     return ssl_aware_request('GET', url, **kwargs)
 
+
 def is_windows_11():
     if sys.platform != "win32":
         return False