Issue #198 Code review: simplify and encapsulate getting/setting file permissions.

Author: JohanKJSchreurs

openeo/rest/auth/config.py

@@ -8,7 +8,6 @@
 
 import json
 import logging
-import platform
 import stat
 from datetime import datetime
 from pathlib import Path
@@ -18,24 +17,40 @@
 from openeo.config import get_user_config_dir, get_user_data_dir
 from openeo.util import rfc3339, deep_get, deep_set
 
-if platform.system() == 'Windows':
+# Use oschmod if it is installed.
+# Otherwise we fall back on Python's chmod implementation.
+# On Windows we have to use oschmod because Python chmod implementation doesn't work on Windows.
+# TODO: Can we use oschmod transparantly on all platforms?
+#       Seems like that is the intention of oschmod, but for now we will play it safe.
+try:
     import oschmod
+except ImportError:
+    oschmod = None
 
 
 _PRIVATE_PERMS = stat.S_IRUSR | stat.S_IWUSR
 
 log = logging.getLogger(__name__)
 
 
-def assert_private_file(path: Path):
-    """Check that given file is only readable by user."""
+def get_file_mode(path: Path):
+    """Get the file permission bits in a way that works on both *nix and Windows platforms."""
+    if oschmod:
+        return oschmod.get_mode(str(path))
+    return path.stat().st_mode
 
-    # use oschmod on Windows
-    # TODO: can we use oschmod for all operating systems, make the code simpler and consitent?
-    if platform.system() == "Windows":
-        mode = oschmod.get_mode(str(path))
+
+def set_file_mode(path: Path, mode):
+    """Set the file permission bits in a way that works on both *nix and Windows platforms."""
+    if oschmod:
+        oschmod.set_mode(str(path), mode=mode)
     else:
-        mode = path.stat().st_mode
+        path.chmod(mode=mode)
+
+
+def assert_private_file(path: Path):
+    """Check that given file is only readable by user."""
+    mode = get_file_mode(path)
     if (mode & stat.S_IRWXG) or (mode & stat.S_IRWXO):
         message = "File {p} could be readable by others: mode {a:o} (expected: {e:o}).".format(
             p=path, a=mode & (stat.S_IRWXU | stat.S_IRWXG | stat.S_IRWXO), e=_PRIVATE_PERMS
@@ -93,12 +108,7 @@ def _write(self, data: dict):
         # TODO: add file locking to avoid race conditions?
         with self._path.open("w", encoding="utf8") as f:
             json.dump(data, f, indent=2)
-        # On Windows we use oschmod because Python chmod implementation doesn't work on Windows.
-        # TODO: can we use oschmod on all platforms? Seems like that is the intention of oschmod.
-        if platform.system() == "Windows":
-            oschmod.set_mode(str(self._path), mode=_PRIVATE_PERMS)
-        else:
-            self._path.chmod(mode=_PRIVATE_PERMS)
+        set_file_mode(self._path, mode=_PRIVATE_PERMS)
         assert_private_file(self._path)
 
     def get(self, *keys, default=None) -> Union[dict, str, int]:

setup.py

@@ -47,8 +47,7 @@
           'xarray>=0.12.3',
           'pandas>0.20.0',
           'deprecated>=1.2.12',
-          'oschmod; sys_platform == "win32"',
-          'pywin32; sys_platform == "win32"',
+          'oschmod>=0.3.12',
       ],
       extras_require={
           "dev": tests_require + [

tests/rest/auth/test_config.py

@@ -1,18 +1,10 @@
 import json
 from unittest import mock
-import platform
 
 import pytest
 
 import openeo.rest.auth.config
-from openeo.rest.auth.config import RefreshTokenStore, AuthConfig, PrivateJsonFile
-
-# TODO: We could simplify the code if we can verify oschmod works transparently on Linux/Mac as well.
-#   I think oschmod does work on Linux and Mac, but since it we use it for a
-#   security sensitive feature, let's make sure there are no problems first.
-#   For now, it is fine to have some Windows specific code.
-if platform.system() == 'Windows':
-    import oschmod
+from openeo.rest.auth.config import RefreshTokenStore, AuthConfig, PrivateJsonFile, get_file_mode
 
 
 class TestPrivateJsonFile:
@@ -39,10 +31,7 @@ def test_permissions(self, tmp_path):
         assert not private.path.exists()
         private.set("foo", "bar", value=42)
         assert private.path.exists()
-        if platform.system() == 'Windows':
-            st_mode = oschmod.get_mode(str(private.path))
-        else:
-            st_mode = private.path.stat().st_mode
+        st_mode = get_file_mode(private.path)
         assert st_mode & 0o777 == 0o600
 
     def test_wrong_permissions(self, tmp_path):
@@ -162,12 +151,7 @@ def test_public_file(self, tmp_path):
     def test_permissions(self, tmp_path):
         r = RefreshTokenStore(path=tmp_path)
         r.set_refresh_token("foo", "bar", "imd6$3cr3t")
-        token_path = (tmp_path / RefreshTokenStore.DEFAULT_FILENAME)
-        if platform.system() == 'Windows':
-            st_mode = oschmod.get_mode(str(token_path))
-        else:
-            token_path.stat().st_mode
-            st_mode = (tmp_path / RefreshTokenStore.DEFAULT_FILENAME).stat().st_mode
+        st_mode = get_file_mode(tmp_path / RefreshTokenStore.DEFAULT_FILENAME)
         assert st_mode & 0o777 == 0o600
 
     def test_get_set_refresh_token(self, tmp_path):

tests/test_config.py

@@ -1,7 +1,6 @@
 import contextlib
 import logging
 import os
-import platform
 import random
 import re
 import textwrap
@@ -160,13 +159,7 @@ def test_get_config_verbose(tmp_path, caplog, capsys, verbose, force_interactive
         config = get_config()
         assert config.get("general.verbose") == verbose
 
-    # On Windows the path in config_path would make the regex invalid because
-    # windows paths contain \ and a regex interprets that as an escape sequence.
-    # Therefore we have to escape the \ to \\ .
-    if platform.system() == "Windows":
-        config_path_escaped = str(config_path).replace("\\", "\\\\")
-    else:
-        config_path_escaped = config_path
-    regex = re.compile(f"Loaded.*config from.*{config_path_escaped}")
+    # re.escape config_path because Windows paths contain "\"
+    regex = re.compile(f"Loaded.*config from.*{re.escape(config_path)}")
     assert regex.search(caplog.text)
     assert bool(regex.search(capsys.readouterr().out)) == on_stdout