Normalize file persistence handling for existing files

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

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

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},
         ))
-