fix: enforce owner-only permissions on credential files

Credentials stored at rest were left world-readable on multi-user hosts:

- TokenManager._get_secure_storage_path() documented its credential dir as
  mode 0o700 but created it via mkdir() with default perms (0o755), leaving
  the Fernet secret.key and encrypted tokens.enc in a traversable dir.
- Settings.dump() persisted tool_repository_password (plaintext) to
  settings.json via open("w"), producing a 0o644 file, and created the
  config dir at 0o755 — despite the sibling token_manager already writing
  secrets atomically at 0o600.

Fixes:
- TokenManager: chmod the credential dir to 0o700 after mkdir (robust against
  umask and pre-existing dirs).
- Settings: write settings.json atomically at 0o600 (mkstemp + chmod +
  os.replace) and chmod the dedicated config dir to 0o700. The /tmp and cwd
  fallback parents are deliberately not chmod'd; the 0o600 file mode protects
  the credential there.

Adds regression tests asserting 0o600 files and 0o700 dirs, and that shared
fallback dirs are not globally tightened.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
This commit is contained in:
Rip&Tear
2026-06-19 14:23:30 +08:00
parent 854c67d21c
commit 750684ca36
4 changed files with 144 additions and 2 deletions

View File

@@ -1,5 +1,8 @@
import json
import os
import shutil
import stat
import sys
import tempfile
import unittest
from datetime import datetime, timedelta
@@ -146,3 +149,55 @@ class TestSettings(unittest.TestCase):
settings = Settings(config_path=self.config_path)
self.assertIsNone(settings.tool_repository_username)
class TestSettingsFilePermissions(unittest.TestCase):
"""Regression tests: credentials in settings.json must not be world-readable."""
def setUp(self):
self.test_dir = Path(tempfile.mkdtemp())
def tearDown(self):
shutil.rmtree(self.test_dir, ignore_errors=True)
@unittest.skipIf(sys.platform == "win32", "POSIX permission semantics")
def test_dump_writes_owner_only_file(self):
config_path = self.test_dir / "settings.json"
old_umask = os.umask(0o022)
try:
settings = Settings(
config_path=config_path, tool_repository_password="hunter2"
)
settings.dump()
finally:
os.umask(old_umask)
mode = stat.S_IMODE(config_path.stat().st_mode)
self.assertEqual(mode, 0o600, f"expected 0o600, got {oct(mode)}")
@unittest.skipIf(sys.platform == "win32", "POSIX permission semantics")
def test_dedicated_config_dir_is_owner_only(self):
config_path = self.test_dir / "crewai" / "settings.json"
old_umask = os.umask(0o022)
try:
Settings(config_path=config_path, tool_repository_username="u")
finally:
os.umask(old_umask)
mode = stat.S_IMODE(config_path.parent.stat().st_mode)
self.assertEqual(mode, 0o700, f"expected 0o700, got {oct(mode)}")
@unittest.skipIf(sys.platform == "win32", "POSIX permission semantics")
def test_shared_fallback_dir_is_not_chmodded(self):
"""The system temp dir (a fallback parent) must never be globally chmod'd."""
from crewai_core.settings import _ensure_dir_mode
tmp_root = Path(tempfile.gettempdir())
before = stat.S_IMODE(tmp_root.stat().st_mode)
_ensure_dir_mode(tmp_root)
after = stat.S_IMODE(tmp_root.stat().st_mode)
self.assertEqual(before, after)
if __name__ == "__main__":
unittest.main()

View File

@@ -1,6 +1,9 @@
"""Tests for TokenManager with atomic file operations."""
import json
import os
import stat
import sys
import tempfile
import unittest
from datetime import datetime, timedelta
@@ -285,5 +288,50 @@ class TestAtomicFileOperations(unittest.TestCase):
tm._delete_secure_file("nonexistent.txt")
class TestSecureStoragePathPermissions(unittest.TestCase):
"""Test that the credential directory is created with restrictive permissions."""
@unittest.skipIf(sys.platform == "win32", "POSIX permission semantics")
def test_storage_path_is_owner_only(self) -> None:
"""The credential directory must be mode 0o700 even under a permissive umask."""
with tempfile.TemporaryDirectory() as base:
old_umask = os.umask(0o022)
try:
with (
patch("crewai_core.token_manager.sys.platform", "linux"),
patch(
"crewai_core.token_manager.os.path.expanduser",
return_value=base,
),
):
storage_path = TokenManager._get_secure_storage_path()
finally:
os.umask(old_umask)
self.assertTrue(storage_path.is_dir())
mode = stat.S_IMODE(storage_path.stat().st_mode)
self.assertEqual(mode, 0o700, f"expected 0o700, got {oct(mode)}")
@unittest.skipIf(sys.platform == "win32", "POSIX permission semantics")
def test_existing_loose_dir_is_tightened(self) -> None:
"""A pre-existing world-traversable directory is corrected to 0o700."""
with tempfile.TemporaryDirectory() as base:
loose = Path(base) / "crewai" / "credentials"
loose.mkdir(parents=True)
loose.chmod(0o755)
with (
patch("crewai_core.token_manager.sys.platform", "linux"),
patch(
"crewai_core.token_manager.os.path.expanduser",
return_value=base,
),
):
storage_path = TokenManager._get_secure_storage_path()
mode = stat.S_IMODE(storage_path.stat().st_mode)
self.assertEqual(mode, 0o700, f"expected 0o700, got {oct(mode)}")
if __name__ == "__main__":
unittest.main()