From 5b697a5919dbb5da4edc6b8cfd533b4c6325acdd Mon Sep 17 00:00:00 2001 From: Dharshan BJ Date: Wed, 28 Oct 2026 16:06:23 -0700 Subject: [PATCH] Normalize file persistence permissions before saving Apply owner read/write mode to existing POSIX cache files before truncation, set the creation mode for marker files, and cover permission error handling and marker reuse. Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> Copilot-Session: 5db15715-2b1f-4a66-b2da-693075d57717 --- msal_extensions/persistence.py | 19 +++++++++--- tests/test_agnostic_backend.py | 29 +++++++++++++++++++ tests/test_persistence.py | 53 +++++++++++++++++++++++++++++++++- 3 files changed, 96 insertions(+), 5 deletions(-) diff --git a/msal_extensions/persistence.py b/msal_extensions/persistence.py index 8056c30..206e459 100644 --- a/msal_extensions/persistence.py +++ b/msal_extensions/persistence.py @@ -134,12 +134,23 @@ def get_location(self): def _open(location): - return os.open(location, os.O_RDWR | os.O_CREAT | os.O_TRUNC, 0o600) - # The 600 seems no-op on NTFS/Windows, and that is fine + descriptor = os.open(location, os.O_RDWR | os.O_CREAT, 0o600) + try: + if os.name == "posix": + # Creation mode does not restrict an existing file, such as a legacy marker. + os.fchmod(descriptor, 0o600) + os.ftruncate(descriptor, 0) + except OSError: + os.close(descriptor) + raise + return descriptor class FilePersistence(BasePersistence): - """A generic persistence, storing data in a plain-text file""" + """A generic persistence, storing data in a plain-text file. + + On POSIX, saving restricts the file permissions to owner read/write. + """ def __init__(self, location): if not location: @@ -185,7 +196,7 @@ def time_last_modified(self): def touch(self): """To touch this file-based persistence without writing content into it""" - Path(self._location).touch() # For os.path.getmtime() to work + Path(self._location).touch(mode=0o600) # For os.path.getmtime() to work def get_location(self): return self._location diff --git a/tests/test_agnostic_backend.py b/tests/test_agnostic_backend.py index d241a2d..22c5f77 100644 --- a/tests/test_agnostic_backend.py +++ b/tests/test_agnostic_backend.py @@ -4,6 +4,8 @@ import tempfile from unittest.mock import patch import sys +import runpy +import stat import msal import pytest @@ -47,3 +49,30 @@ def test_file_not_found_error_is_not_raised(): cache = PersistedTokenCache(persistence) # An exception raised here will fail the test case as it is supposed to be a NO-OP cache.find('') + +@pytest.mark.skipif(os.name != "posix", reason="Requires POSIX file permissions") +def test_sample_plaintext_fallback_with_existing_marker(temp_location): + marker = FilePersistence(temp_location) + marker.touch() + os.chmod(temp_location, 0o644) # A marker left by an older native-store persistence. + sample = os.path.join( + os.path.dirname(os.path.dirname(__file__)), "sample", "token_cache_sample.py") + with patch("msal_extensions.build_encrypted_persistence", return_value=marker) as builder: + build_persistence = runpy.run_path(sample)["build_persistence"] + builder.side_effect = RuntimeError("Native store unavailable") + with pytest.raises(RuntimeError, match="Native store unavailable"): + build_persistence(temp_location) + persistence = build_persistence(temp_location, fallback_to_plaintext=True) + + cache = PersistedTokenCache(persistence) + cache.add({ + "client_id": "test-client", + "scope": ["test-scope"], + "token_endpoint": "https://login.microsoftonline.com/test-tenant/oauth2/v2.0/token", + "response": {"refresh_token": "synthetic-refresh-token"}, + }) + + assert not cache.is_encrypted + tokens = json.loads(persistence.load())["RefreshToken"].values() + assert any(token["secret"] == "synthetic-refresh-token" for token in tokens) + assert stat.S_IMODE(os.stat(temp_location).st_mode) == 0o600 diff --git a/tests/test_persistence.py b/tests/test_persistence.py index d451225..a5e62af 100644 --- a/tests/test_persistence.py +++ b/tests/test_persistence.py @@ -3,6 +3,9 @@ import shutil import tempfile import logging +import errno +import stat +from unittest.mock import patch import pytest @@ -44,6 +47,55 @@ def test_file_persistence(temp_location): def test_nonexistent_file_persistence(temp_location): _test_nonexistent_persistence(FilePersistence(temp_location)) +@pytest.mark.skipif(os.name != "posix", reason="Requires POSIX file permissions") +@pytest.mark.parametrize("mode", [0o644, 0o666]) +def test_file_persistence_restricts_existing_permissions(temp_location, mode): + with open(temp_location, "w") as handle: + handle.write("previous content") + os.chmod(temp_location, mode) + + persistence = FilePersistence(temp_location) + persistence.save("new") + + assert persistence.load() == "new" + assert stat.S_IMODE(os.stat(temp_location).st_mode) == 0o600 + +@pytest.mark.skipif(os.name != "posix", reason="Requires POSIX file permissions") +def test_file_persistence_permission_failure_preserves_content(temp_location): + with open(temp_location, "w") as handle: + handle.write("previous content") + os.chmod(temp_location, 0o644) + persistence = FilePersistence(temp_location) + + with patch("msal_extensions.persistence.os.fchmod", + side_effect=OSError(errno.EPERM, "Permission denied")) as chmod: + with pytest.raises(OSError) as error: + persistence.save("new secret") + + assert error.value.errno == errno.EPERM + assert persistence.load() == "previous content" + assert stat.S_IMODE(os.stat(temp_location).st_mode) == 0o644 + with pytest.raises(OSError) as closed: + os.fstat(chmod.call_args[0][0]) + assert closed.value.errno == errno.EBADF + +@pytest.mark.skipif(os.name != "posix", reason="Requires POSIX file permissions") +def test_file_persistence_touch_creates_private_marker(temp_location): + persistence = FilePersistence(temp_location) + previous_umask = os.umask(0o022) + try: + persistence.touch() + finally: + os.umask(previous_umask) + + assert persistence.load() == "" + assert stat.S_IMODE(os.stat(temp_location).st_mode) == 0o600 + persistence.save("existing content") + os.utime(temp_location, (1, 1)) + persistence.touch() + assert persistence.load() == "existing content" + assert persistence.time_last_modified() > 1 + @pytest.mark.skipif( not sys.platform.startswith('win'), reason="Requires Windows Desktop") @@ -92,4 +144,3 @@ def test_nonexistent_libsecret_persistence(temp_location): random_schema_name, {"my_attr_1": random_value, "my_attr_2": random_value}, )) -