add verification of directory permissions too

Author: sfc-gh-pczajka

src/snowflake/connector/crl_cache.py

@@ -227,10 +227,19 @@ def __init__(
         self._ensure_cache_directory_exists()
 
     def _ensure_cache_directory_exists(self) -> None:
-        """Create the cache directory if it doesn't exist."""
+        """Create the cache directory if it doesn't exist with secure permissions."""
         try:
-            self._cache_dir.mkdir(parents=True, exist_ok=True)
+            # Create directory with secure permissions (owner read/write/execute only)
+            self._cache_dir.mkdir(parents=True, exist_ok=True, mode=0o700)
+
+            # Verify directory permissions (if it already existed)
+            if not self._unsafe_skip_file_permissions_check:
+                self._check_permissions(self._cache_dir, "directory", "0o700")
+
             logger.debug(f"Cache directory created/verified: {self._cache_dir}")
+        except PermissionError:
+            # Re-raise permission errors as-is
+            raise
         except OSError as e:
             raise OSError(f"Failed to create cache directory {self._cache_dir}: {e}")
 
@@ -256,38 +265,42 @@ 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:
+    def _check_permissions(
+        self, path: Path, resource_type: str, expected_perms: str
+    ) -> None:
         """
-        Check that the CRL cache file has secure permissions (owner-only access).
+        Check that a CRL cache resource 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
+            path: Path to the resource (file or directory) to check
+            resource_type: Description of the resource type (e.g., "file", "directory")
+            expected_perms: Description of expected permissions (e.g., "0o600 or 0o400", "0o700")
 
         Raises:
-            PermissionError: If file permissions are too wide or file has wrong owner
+            PermissionError: If resource permissions are too wide
         """
         # Skip permission checks on Windows as they work differently
         if IS_WINDOWS:
             return
 
         try:
-            stat_info = file_path.stat()
+            stat_info = path.stat()
             actual_permissions = stat.S_IMODE(stat_info.st_mode)
 
-            # Check that file is readable/writable only by owner (0o600 or more restrictive)
+            # Check that resource is accessible only by owner (no group/other permissions)
             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)."
+                    f"CRL cache {resource_type} {path} has insecure permissions: {oct(actual_permissions)}. "
+                    f"{resource_type.capitalize()} must be accessible only by the owner ({expected_perms})."
                 )
 
         except FileNotFoundError:
-            # File doesn't exist yet, this is fine
+            # Resource doesn't exist yet, this is fine
             pass
 
     def get(self, crl_url: str) -> CRLCacheEntry | None:
@@ -308,7 +321,7 @@ def get(self, crl_url: str) -> CRLCacheEntry | None:
 
                     # Check file permissions before reading
                     if not self._unsafe_skip_file_permissions_check:
-                        self._check_file_permissions(crl_file_path)
+                        self._check_permissions(crl_file_path, "file", "0o600 or 0o400")
                     else:
                         logger.warning(
                             f"Skipping file permissions check for {crl_file_path}"

test/unit/test_crl_cache.py

@@ -707,3 +707,74 @@ def test_file_cache_permission_validation(
             # Secure permissions or checks disabled - should read successfully
             assert result is not None
             assert result.crl is not None
+
+
+@pytest.mark.skipif(
+    IS_WINDOWS, reason="File permission checks not applicable on Windows"
+)
+def test_cache_directory_created_with_secure_permissions():
+    """Test that cache directory is created with owner-only permissions (0o700)"""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        cache_dir = Path(temp_dir) / "crl_cache"
+        # initialize the cache directory
+        CRLFileCache(cache_dir, timedelta(hours=1))
+
+        # Check directory permissions
+        dir_stat = cache_dir.stat()
+        permissions = stat.S_IMODE(dir_stat.st_mode)
+
+        # Should be 0o700 (read/write/execute for owner only)
+        assert permissions == 0o700, f"Expected 0o700, got {oct(permissions)}"
+
+
+@pytest.mark.skipif(
+    IS_WINDOWS, reason="File permission checks not applicable on Windows"
+)
+@pytest.mark.parametrize(
+    "chmod_permissions,skip_check,should_raise",
+    [
+        (0o755, False, True),  # rwxr-xr-x (world-readable) with checks - should raise
+        (0o750, False, True),  # rwxr-x--- (group-readable) with checks - should raise
+        (0o700, False, False),  # rwx------ (owner only) with checks - should not raise
+        (
+            0o755,
+            True,
+            False,
+        ),  # rwxr-xr-x (world-readable) without checks - should not raise
+    ],
+    ids=[
+        "world-readable-dir",
+        "group-readable-dir",
+        "owner-only-dir",
+        "skip-check-dir",
+    ],
+)
+def test_cache_directory_permission_validation(
+    chmod_permissions, skip_check, should_raise
+):
+    """Test that cache directory permissions are validated correctly"""
+    with tempfile.TemporaryDirectory() as temp_dir:
+        cache_dir = Path(temp_dir) / "crl_cache"
+        cache_dir.mkdir(mode=0o700)
+
+        # Change directory permissions before creating cache
+        os.chmod(cache_dir, chmod_permissions)
+
+        if should_raise:
+            # Should raise PermissionError for insecure directory
+            with pytest.raises(
+                PermissionError, match="cache directory.*insecure permissions"
+            ):
+                CRLFileCache(
+                    cache_dir,
+                    timedelta(hours=1),
+                    unsafe_skip_file_permissions_check=skip_check,
+                )
+        else:
+            # Should succeed with secure permissions or when checks are disabled
+            cache = CRLFileCache(
+                cache_dir,
+                timedelta(hours=1),
+                unsafe_skip_file_permissions_check=skip_check,
+            )
+            assert cache is not None