SNOW-2127911 Add unsafe_ignore_permission_check flag which turns off permission check validation on unix systems (#2430)

Author: sfc-gh-pczajka

DESCRIPTION.md

@@ -14,6 +14,7 @@ Source code is also available at: https://github.com/snowflakedb/snowflake-conne
   - Added in-band HTTP exception telemetry.
   - Fixed a bug where timezoned timestamps fetched as pandas.DataFrame or pyarrow.Table would overflow for the sake of unnecessary precision. In the case where an overflow cannot be prevented a clear error will be raised now.
   - Fix OAuth authenticator values.
+  - Add `unsafe_skip_file_permissions_check` flag to skip file permissions check on cache and config.
 
 - v3.16.0(July 04,2025)
   - Bumped numpy dependency from <2.1.0 to <=2.2.4.

src/snowflake/connector/auth/_auth.py

@@ -542,7 +542,9 @@ def _delete_temporary_credential(
 
     def get_token_cache(self) -> TokenCache:
         if self._token_cache is None:
-            self._token_cache = TokenCache.make()
+            self._token_cache = TokenCache.make(
+                skip_file_permissions_check=self._rest._connection._unsafe_skip_file_permissions_check
+            )
         return self._token_cache
 
 

src/snowflake/connector/config_manager.py

@@ -295,6 +295,7 @@ def _sub_parsers(self) -> dict[str, ConfigManager]:
 
     def read_config(
         self,
+        skip_file_permissions_check: bool = False,
     ) -> None:
         """Read and cache config file contents.
 
@@ -310,8 +311,11 @@ def read_config(
         read_config_file = tomlkit.TOMLDocument()
 
         # Read in all of the config slices
+        config_slice_options = ConfigSliceOptions(
+            check_permissions=not skip_file_permissions_check
+        )
         for filep, sliceoptions, section in itertools.chain(
-            ((self.file_path, ConfigSliceOptions(), None),),
+            ((self.file_path, config_slice_options, None),),
             self._slices,
         ):
             if sliceoptions.only_in_slice:

src/snowflake/connector/connection.py

@@ -377,6 +377,10 @@ def _get_private_bytes_from_file(
         False,
         bool,
     ),  # SNOW-1944208: add unsafe write flag
+    "unsafe_skip_file_permissions_check": (
+        False,
+        bool,
+    ),  # SNOW-2127911: add flag to opt-out file permissions check
     _VARIABLE_NAME_SERVER_DOP_CAP_FOR_FILE_TRANSFER: (
         _DEFAULT_VALUE_SERVER_DOP_CAP_FOR_FILE_TRANSFER,  # default value
         int,  # type
@@ -489,8 +493,13 @@ def __init__(
         If overwriting values from the default connection is desirable, supply
         the name explicitly.
         """
+        self._unsafe_skip_file_permissions_check = kwargs.get(
+            "unsafe_skip_file_permissions_check", False
+        )
         # initiate easy logging during every connection
-        easy_logging = EasyLoggingConfigPython()
+        easy_logging = EasyLoggingConfigPython(
+            skip_config_file_permissions_check=self._unsafe_skip_file_permissions_check
+        )
         easy_logging.create_log()
         self._lock_sequence_counter = Lock()
         self.sequence_counter = 0
@@ -549,7 +558,9 @@ def __init__(
             for i, s in enumerate(CONFIG_MANAGER._slices):
                 if s.section == "connections":
                     CONFIG_MANAGER._slices[i] = s._replace(path=connections_file_path)
-                    CONFIG_MANAGER.read_config()
+                    CONFIG_MANAGER.read_config(
+                        skip_file_permissions_check=self._unsafe_skip_file_permissions_check
+                    )
                     break
         if connection_name is not None:
             connections = CONFIG_MANAGER["connections"]

src/snowflake/connector/log_configuration.py

@@ -12,14 +12,16 @@
 
 
 class EasyLoggingConfigPython:
-    def __init__(self):
+    def __init__(self, skip_config_file_permissions_check: bool = False):
         self.path: str | None = None
         self.level: str | None = None
         self.save_logs: bool = False
-        self.parse_config_file()
+        self.parse_config_file(skip_config_file_permissions_check)
 
-    def parse_config_file(self):
-        CONFIG_MANAGER.read_config()
+    def parse_config_file(self, skip_config_file_permissions_check: bool = False):
+        CONFIG_MANAGER.read_config(
+            skip_file_permissions_check=skip_config_file_permissions_check
+        )
         data = CONFIG_MANAGER.conf_file_cache
         if log := data.get("log"):
             self.save_logs = log.get("save_logs", False)

src/snowflake/connector/token_cache.py

@@ -58,7 +58,7 @@ def _warn(warning: str) -> None:
 
 class TokenCache(ABC):
     @staticmethod
-    def make() -> TokenCache:
+    def make(skip_file_permissions_check: bool = False) -> TokenCache:
         if IS_MACOS or IS_WINDOWS:
             if not installed_keyring:
                 _warn(
@@ -71,7 +71,7 @@ def make() -> TokenCache:
             return KeyringTokenCache()
 
         if IS_LINUX:
-            cache = FileTokenCache.make()
+            cache = FileTokenCache.make(skip_file_permissions_check)
             if cache:
                 return cache
             else:
@@ -128,23 +128,30 @@ class _CacheFileWriteError(_FileTokenCacheError):
 
 class FileTokenCache(TokenCache):
     @staticmethod
-    def make() -> FileTokenCache | None:
-        cache_dir = FileTokenCache.find_cache_dir()
+    def make(skip_file_permissions_check: bool = False) -> FileTokenCache | None:
+        cache_dir = FileTokenCache.find_cache_dir(skip_file_permissions_check)
         if cache_dir is None:
             logging.getLogger(__name__).debug(
                 "Failed to find suitable cache directory for token cache. File based token cache initialization failed."
             )
             return None
         else:
-            return FileTokenCache(cache_dir)
+            return FileTokenCache(
+                cache_dir, skip_file_permissions_check=skip_file_permissions_check
+            )
 
-    def __init__(self, cache_dir: Path) -> None:
+    def __init__(
+        self, cache_dir: Path, skip_file_permissions_check: bool = False
+    ) -> None:
         self.logger = logging.getLogger(__name__)
         self.cache_dir: Path = cache_dir
+        self._skip_file_permissions_check = skip_file_permissions_check
 
     def store(self, key: TokenKey, token: str) -> None:
         try:
-            FileTokenCache.validate_cache_dir(self.cache_dir)
+            FileTokenCache.validate_cache_dir(
+                self.cache_dir, self._skip_file_permissions_check
+            )
             with FileLock(self.lock_file()):
                 cache = self._read_cache_file()
                 cache["tokens"][key.hash_key()] = token
@@ -158,7 +165,9 @@ def store(self, key: TokenKey, token: str) -> None:
 
     def retrieve(self, key: TokenKey) -> str | None:
         try:
-            FileTokenCache.validate_cache_dir(self.cache_dir)
+            FileTokenCache.validate_cache_dir(
+                self.cache_dir, self._skip_file_permissions_check
+            )
             with FileLock(self.lock_file()):
                 cache = self._read_cache_file()
                 token = cache["tokens"].get(key.hash_key(), None)
@@ -178,7 +187,9 @@ def retrieve(self, key: TokenKey) -> str | None:
 
     def remove(self, key: TokenKey) -> None:
         try:
-            FileTokenCache.validate_cache_dir(self.cache_dir)
+            FileTokenCache.validate_cache_dir(
+                self.cache_dir, self._skip_file_permissions_check
+            )
             with FileLock(self.lock_file()):
                 cache = self._read_cache_file()
                 cache["tokens"].pop(key.hash_key(), None)
@@ -201,7 +212,8 @@ def _read_cache_file(self) -> dict[str, dict[str, Any]]:
         json_data = {"tokens": {}}
         try:
             fd = os.open(self.cache_file(), os.O_RDONLY)
-            self._ensure_permissions(fd, 0o600)
+            if not self._skip_file_permissions_check:
+                self._ensure_permissions(fd, 0o600)
             size = os.lseek(fd, 0, os.SEEK_END)
             os.lseek(fd, 0, os.SEEK_SET)
             data = os.read(fd, size)
@@ -234,7 +246,8 @@ def _write_cache_file(self, json_data: dict):
             fd = os.open(
                 self.cache_file(), os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600
             )
-            self._ensure_permissions(fd, 0o600)
+            if not self._skip_file_permissions_check:
+                self._ensure_permissions(fd, 0o600)
             os.write(fd, codecs.encode(json.dumps(json_data), "utf-8"))
             return json_data
         except OSError as e:
@@ -244,7 +257,7 @@ def _write_cache_file(self, json_data: dict):
                 os.close(fd)
 
     @staticmethod
-    def find_cache_dir() -> Path | None:
+    def find_cache_dir(skip_file_permissions_check: bool = False) -> Path | None:
         def lookup_env_dir(env_var: str, subpath_segments: list[str]) -> Path | None:
             env_val = os.getenv(env_var)
             if env_val is None:
@@ -276,10 +289,12 @@ def lookup_env_dir(env_var: str, subpath_segments: list[str]) -> Path | None:
                 directory.mkdir(exist_ok=True, mode=0o700)
 
             try:
-                FileTokenCache.validate_cache_dir(directory)
+                FileTokenCache.validate_cache_dir(
+                    directory, skip_file_permissions_check
+                )
                 return directory
             except _FileTokenCacheError as e:
-                logger.debug(
+                _warn(
                     f"Cache directory validation failed for {str(directory)} due to error '{e}'. Skipping it in cache directory lookup."
                 )
                 return None
@@ -298,7 +313,9 @@ def lookup_env_dir(env_var: str, subpath_segments: list[str]) -> Path | None:
         return None
 
     @staticmethod
-    def validate_cache_dir(cache_dir: Path | None) -> None:
+    def validate_cache_dir(
+        cache_dir: Path | None, skip_file_permissions_check: bool = False
+    ) -> None:
         try:
             statinfo = cache_dir.stat()
 
@@ -308,17 +325,18 @@ def validate_cache_dir(cache_dir: Path | None) -> None:
             if not stat.S_ISDIR(statinfo.st_mode):
                 raise _InvalidCacheDirError(f"Cache dir {cache_dir} is not a directory")
 
-            permissions = stat.S_IMODE(statinfo.st_mode)
-            if permissions != 0o700:
-                raise _PermissionsTooWideError(
-                    f"Cache dir {cache_dir} has incorrect permissions. {permissions:o} != 0700"
-                )
+            if not skip_file_permissions_check:
+                permissions = stat.S_IMODE(statinfo.st_mode)
+                if permissions != 0o700:
+                    raise _PermissionsTooWideError(
+                        f"Cache dir {cache_dir} has incorrect permissions. {permissions:o} != 0700"
+                    )
 
-            euid = os.geteuid()
-            if statinfo.st_uid != euid:
-                raise _OwnershipError(
-                    f"Cache dir {cache_dir} has incorrect owner. {euid} != {statinfo.st_uid}"
-                )
+                euid = os.geteuid()
+                if statinfo.st_uid != euid:
+                    raise _OwnershipError(
+                        f"Cache dir {cache_dir} has incorrect owner. {euid} != {statinfo.st_uid}"
+                    )
 
         except FileNotFoundError:
             raise _CacheDirNotFoundError(

test/integ/test_connection.py

@@ -18,6 +18,7 @@
 
 import snowflake.connector
 from snowflake.connector import DatabaseError, OperationalError, ProgrammingError
+from snowflake.connector.compat import IS_WINDOWS
 from snowflake.connector.connection import (
     DEFAULT_CLIENT_PREFETCH_THREADS,
     SnowflakeConnection,
@@ -1639,3 +1640,56 @@ def test_file_utils_sanity_check():
     conn = create_connection("default")
     assert hasattr(conn._file_operation_parser, "parse_file_operation")
     assert hasattr(conn._stream_downloader, "download_as_stream")
+
+
+@pytest.mark.skipolddriver
+@pytest.mark.skipif(IS_WINDOWS, reason="chmod doesn't work on Windows")
+def test_unsafe_skip_file_permissions_check_skips_config_permissions_check(
+    db_parameters, tmp_path
+):
+    """Test that unsafe_skip_file_permissions_check flag bypasses permission checks on config files."""
+    # Write config file and set unsafe permissions (readable by others)
+    tmp_config_file = tmp_path / "config.toml"
+    tmp_config_file.write_text("[log]\n" "save_logs = false\n" 'level = "INFO"\n')
+    tmp_config_file.chmod(stat.S_IRUSR | stat.S_IWUSR | stat.S_IROTH)
+
+    def _run_select_1(unsafe_skip_file_permissions_check: bool):
+        warnings.simplefilter("always")
+        # Connect directly with db_parameters, using custom config file path
+        # We need to modify CONFIG_MANAGER to point to our test file
+        from snowflake.connector.config_manager import CONFIG_MANAGER
+
+        original_file_path = CONFIG_MANAGER.file_path
+        try:
+            CONFIG_MANAGER.file_path = tmp_config_file
+            CONFIG_MANAGER.conf_file_cache = None  # Force re-read
+            with snowflake.connector.connect(
+                **db_parameters,
+                unsafe_skip_file_permissions_check=unsafe_skip_file_permissions_check,
+            ) as conn:
+                with conn.cursor() as cur:
+                    result = cur.execute("select 1;").fetchall()
+                    assert result == [(1,)]
+        finally:
+            CONFIG_MANAGER.file_path = original_file_path
+            CONFIG_MANAGER.conf_file_cache = None
+
+    # Without the flag - should trigger permission warnings
+    with warnings.catch_warnings(record=True) as warning_list:
+        _run_select_1(unsafe_skip_file_permissions_check=False)
+    permission_warnings = [
+        w for w in warning_list if "Bad owner or permissions" in str(w.message)
+    ]
+    assert (
+        len(permission_warnings) > 0
+    ), "Expected permission warning when unsafe_skip_file_permissions_check=False"
+
+    # With the flag - should bypass permission checks and not show warnings
+    with warnings.catch_warnings(record=True) as warning_list:
+        _run_select_1(unsafe_skip_file_permissions_check=True)
+    permission_warnings = [
+        w for w in warning_list if "Bad owner or permissions" in str(w.message)
+    ]
+    assert (
+        len(permission_warnings) == 0
+    ), "Expected no permission warning when unsafe_skip_file_permissions_check=True"