file permissions check

Author: sfc-gh-pczajka

src/snowflake/connector/connection.py

@@ -797,6 +797,13 @@ def crl_download_max_size(self) -> int | None:
             return self._crl_download_max_size
         return self._crl_config.crl_download_max_size
 
+    @property
+    def skip_file_permissions_check(self) -> bool | None:
+        """Whether to skip file permission checks for CRL cache files."""
+        if not self._crl_config:
+            return self._skip_file_permissions_check
+        return self._crl_config.skip_file_permissions_check
+
     @property
     def session_id(self) -> int:
         return self._session_id

src/snowflake/connector/crl.py

@@ -65,6 +65,7 @@ class CRLConfig:
     crl_cache_cleanup_interval_hours: int = 1
     crl_cache_start_cleanup: bool = False
     crl_download_max_size: int = 200 * 1024 * 1024  # 200 MB
+    unsafe_skip_file_permissions_check: bool = False
 
     @classmethod
     def from_connection(cls, sf_connection) -> CRLConfig:
@@ -164,6 +165,10 @@ def from_connection(cls, sf_connection) -> CRLConfig:
             if sf_connection.crl_download_max_size is None
             else int(sf_connection.crl_download_max_size)
         )
+        # Use the existing unsafe_skip_file_permissions_check flag from connection
+        unsafe_skip_file_permissions_check = bool(
+            sf_connection._unsafe_skip_file_permissions_check
+        )
 
         return cls(
             cert_revocation_check_mode=cert_revocation_check_mode,
@@ -178,6 +183,7 @@ def from_connection(cls, sf_connection) -> CRLConfig:
             crl_cache_cleanup_interval_hours=crl_cache_cleanup_interval_hours,
             crl_cache_start_cleanup=crl_cache_start_cleanup,
             crl_download_max_size=crl_download_max_size,
+            unsafe_skip_file_permissions_check=unsafe_skip_file_permissions_check,
         )
 
 
@@ -246,7 +252,9 @@ def from_config(
             if config.enable_crl_file_cache:
                 removal_delay = timedelta(days=config.crl_cache_removal_delay_days)
                 file_cache = CRLCacheFactory.get_file_cache(
-                    cache_dir=config.crl_cache_dir, removal_delay=removal_delay
+                    cache_dir=config.crl_cache_dir,
+                    removal_delay=removal_delay,
+                    unsafe_skip_file_permissions_check=config.unsafe_skip_file_permissions_check,
                 )
             else:
                 from snowflake.connector.crl_cache import NoopCRLCache

src/snowflake/connector/crl_cache.py

@@ -6,6 +6,7 @@
 import logging
 import os
 import platform
+import stat
 import threading
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
@@ -17,6 +18,8 @@
 from cryptography.hazmat.primitives import serialization
 from filelock import BaseFileLock, FileLock
 
+from .compat import IS_WINDOWS
+
 logger = logging.getLogger(__name__)
 
 
@@ -200,21 +203,26 @@ class CRLFileCache(CRLCache):
     """
 
     def __init__(
-        self, cache_dir: Path | None = None, removal_delay: timedelta | None = None
+        self,
+        cache_dir: Path | None = None,
+        removal_delay: timedelta | None = None,
+        unsafe_skip_file_permissions_check: bool = False,
     ):
         """
         Initialize the file cache.
 
         Args:
             cache_dir: Directory to store cached CRLs
             removal_delay: How long to wait before removing expired files
+            unsafe_skip_file_permissions_check: Skip file permission validation for security
 
         Raises:
             OSError: If cache directory cannot be created
         """
         self._cache_file_lock_timeout = 5.0
         self._cache_dir = cache_dir or _get_default_crl_cache_path()
         self._removal_delay = removal_delay or timedelta(days=7)
+        self._unsafe_skip_file_permissions_check = unsafe_skip_file_permissions_check
 
         self._ensure_cache_directory_exists()
 
@@ -248,6 +256,40 @@ def _get_crl_file_lock(self, crl_cache_file: Path) -> BaseFileLock:
             timeout=self._cache_file_lock_timeout,
         )
 
+    def _check_file_permissions(self, file_path: Path) -> None:
+        """
+        Check that the CRL cache file has secure permissions (owner-only access).
+
+        Note: This check is only performed on Unix-like systems. Windows file
+        permissions work differently and are not checked.
+
+        Args:
+            file_path: Path to the file to check
+
+        Raises:
+            PermissionError: If file permissions are too wide or file has wrong owner
+        """
+        # Skip permission checks on Windows as they work differently
+        if IS_WINDOWS:
+            return
+
+        try:
+            stat_info = file_path.stat()
+            actual_permissions = stat.S_IMODE(stat_info.st_mode)
+
+            # Check that file is readable/writable only by owner (0o600 or more restrictive)
+            if (
+                actual_permissions & 0o077 != 0
+            ):  # Check if group or others have any permission
+                raise PermissionError(
+                    f"CRL cache file {file_path} has insecure permissions: {oct(actual_permissions)}. "
+                    f"File must be accessible only by the owner (e.g., 0o600 or 0o400)."
+                )
+
+        except FileNotFoundError:
+            # File doesn't exist yet, this is fine
+            pass
+
     def get(self, crl_url: str) -> CRLCacheEntry | None:
         """
         Get a CRL cache entry from disk.
@@ -264,6 +306,14 @@ def get(self, crl_url: str) -> CRLCacheEntry | None:
                 if crl_file_path.exists():
                     logger.debug(f"Found CRL on disk for {crl_file_path}")
 
+                    # Check file permissions before reading
+                    if not self._unsafe_skip_file_permissions_check:
+                        self._check_file_permissions(crl_file_path)
+                    else:
+                        logger.warning(
+                            f"Skipping file permissions check for {crl_file_path}"
+                        )
+
                     # Get file modification time as download time
                     stat_info = crl_file_path.stat()
                     download_time = datetime.fromtimestamp(
@@ -277,6 +327,11 @@ def get(self, crl_url: str) -> CRLCacheEntry | None:
                     crl = x509.load_der_x509_crl(crl_data, backend=default_backend())
                     return CRLCacheEntry(crl, download_time)
 
+            except PermissionError as e:
+                logger.error(
+                    f"Permission error reading CRL from disk cache for {crl_url}: {e}"
+                )
+                return None
             except Exception as e:
                 logger.warning(f"Failed to read CRL from disk cache for {crl_url}: {e}")
 
@@ -296,9 +351,15 @@ def put(self, crl_url: str, entry: CRLCacheEntry) -> None:
                 # Serialize the CRL to DER format
                 crl_data = entry.crl.public_bytes(serialization.Encoding.DER)
 
-                # Write to file
-                with open(crl_file_path, "wb") as f:
-                    f.write(crl_data)
+                # Write to file with secure permissions (owner read/write only)
+                # Using os.open with 0o600 ensures the file is created with secure permissions
+                fd = os.open(
+                    crl_file_path, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600
+                )
+                try:
+                    os.write(fd, crl_data)
+                finally:
+                    os.close(fd)
 
                 # Set file modification time to download time
                 download_timestamp = entry.download_time.timestamp()
@@ -453,25 +514,34 @@ def get_memory_cache(cls, cache_validity_time: timedelta) -> CRLInMemoryCache:
 
     @classmethod
     def get_file_cache(
-        cls, cache_dir: Path | None = None, removal_delay: timedelta | None = None
+        cls,
+        cache_dir: Path | None = None,
+        removal_delay: timedelta | None = None,
+        unsafe_skip_file_permissions_check: bool = False,
     ) -> CRLFileCache:
         """
         Get or create a singleton CRLFileCache instance.
 
         Args:
             cache_dir: Directory to store cached CRLs
             removal_delay: How long to wait before removing expired files
+            unsafe_skip_file_permissions_check: Skip file permission validation for security
 
         Returns:
             The singleton CRLFileCache instance
         """
         with cls._instance_lock:
             if cls._file_cache_instance is None:
-                cls._file_cache_instance = CRLFileCache(cache_dir, removal_delay)
+                cls._file_cache_instance = CRLFileCache(
+                    cache_dir, removal_delay, unsafe_skip_file_permissions_check
+                )
             else:
                 # Check if parameters differ from existing instance
                 existing_cache_dir = cls._file_cache_instance._cache_dir
                 existing_removal_delay = cls._file_cache_instance._removal_delay
+                existing_skip_check = (
+                    cls._file_cache_instance._unsafe_skip_file_permissions_check
+                )
                 requested_cache_dir = cache_dir or _get_default_crl_cache_path()
                 requested_removal_delay = removal_delay or timedelta(days=7)
 
@@ -485,6 +555,11 @@ def get_file_cache(
                         f"CRLs file cache has already been initialized with removal delay of {existing_removal_delay}, "
                         f"ignoring new removal delay of {requested_removal_delay}"
                     )
+                if existing_skip_check != unsafe_skip_file_permissions_check:
+                    logger.warning(
+                        f"CRLs file cache has already been initialized with unsafe_skip_file_permissions_check={existing_skip_check}, "
+                        f"ignoring new value {unsafe_skip_file_permissions_check}"
+                    )
             return cls._file_cache_instance
 
     @classmethod

test/unit/test_crl_cache.py

@@ -618,3 +618,92 @@ def test_atexit_handler_error_handling(cache_factory):
             cache_factory._atexit_cleanup_handler()
         except Exception as e:
             pytest.fail(f"Atexit handler should not raise exceptions: {e}")
+
+
+# File permission tests
+import os
+import stat
+
+from snowflake.connector.compat import IS_WINDOWS
+
+
+@pytest.mark.skipif(
+    IS_WINDOWS, reason="File permission checks not applicable on Windows"
+)
+def test_file_cache_creates_files_with_secure_permissions(crl, download_time):
+    """Test that CRL cache files are created with owner-only permissions (0o600)"""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        cache = CRLFileCache(Path(temp_dir), timedelta(hours=1))
+        crl_url = "http://test.com/secure_crl"
+        entry = CRLCacheEntry(crl, download_time)
+
+        cache.put(crl_url, entry)
+
+        # Get the file path and check its permissions
+        file_path = cache._get_crl_file_path(crl_url)
+        file_stat = file_path.stat()
+        permissions = stat.S_IMODE(file_stat.st_mode)
+
+        # Should be 0o600 (read/write for owner only)
+        assert permissions == 0o600, f"Expected 0o600, got {oct(permissions)}"
+
+
+@pytest.mark.skipif(
+    IS_WINDOWS, reason="File permission checks not applicable on Windows"
+)
+@pytest.mark.parametrize(
+    "chmod_permissions,skip_check,result_should_be_none",
+    [
+        (
+            0o644,
+            False,
+            True,
+        ),  # rw-r--r-- (world-readable) with checks - should be rejected
+        (
+            0o640,
+            False,
+            True,
+        ),  # rw-r----- (group-readable) with checks - should be rejected
+        (
+            0o400,
+            False,
+            False,
+        ),  # r-------- (owner read-only) with checks - should be accepted
+        (
+            0o644,
+            True,
+            False,
+        ),  # rw-r--r-- (world-readable) without checks - should be accepted
+    ],
+    ids=["world-readable", "group-readable", "owner-read-only", "skip-check-flag"],
+)
+def test_file_cache_permission_validation(
+    crl, download_time, chmod_permissions, skip_check, result_should_be_none
+):
+    """Test that file cache validates file permissions correctly and respects skip flag"""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        cache = CRLFileCache(
+            Path(temp_dir),
+            timedelta(hours=1),
+            unsafe_skip_file_permissions_check=skip_check,
+        )
+        crl_url = "http://test.com/test_crl"
+        entry = CRLCacheEntry(crl, download_time)
+
+        # Write the file first
+        cache.put(crl_url, entry)
+
+        # Change file permissions
+        file_path = cache._get_crl_file_path(crl_url)
+        os.chmod(file_path, chmod_permissions)
+
+        # Try to read the file
+        result = cache.get(crl_url)
+
+        if result_should_be_none:
+            # Insecure permissions - should return None
+            assert result is None
+        else:
+            # Secure permissions or checks disabled - should read successfully
+            assert result is not None
+            assert result.crl is not None