Updated pr.

Author: lu-ohai

ads/common/auth.py

@@ -16,8 +16,7 @@
 
 import ads.telemetry
 import oci
-from oci_cli import cli_util
-from ads.common import logger
+from ads.common import logger, utils
 from ads.common.decorator.deprecate import deprecated
 from ads.common.extended_enum import ExtendedEnumMeta
 from oci.config import DEFAULT_LOCATION  # "~/.oci/config"
@@ -877,7 +876,7 @@ def _validate_and_refresh_token(self, configuration: Dict[str, Any]):
 
     def _refresh_security_token(self, configuration: Dict[str, Any]):
         """Refreshes security token. The logic is mainly taken reference from:
-        https://github.com/oracle/oci-cli/blob/9a0978344950d7b7c24a688892f24968dce20ad3/src/oci_cli/cli_session.py#L152
+        https://github.com/oracle/oci-cli/blob/master/src/oci_cli/cli_session.py#L152
 
         Parameters
         ----------
@@ -919,7 +918,7 @@ def _refresh_security_token(self, configuration: Dict[str, Any]):
             refreshed_token = json.loads(response.content.decode('UTF-8'))['token']
             with open(expanded_security_token_location, 'w') as security_token_file:
                 security_token_file.write(refreshed_token)
-            cli_util.apply_user_only_access_permissions(expanded_security_token_location)
+            utils.apply_user_only_access_permissions(expanded_security_token_location)
             logger.info("Successfully refreshed token")
         elif response.status_code == 401:
             raise SecurityTokenError(

ads/common/utils.py

@@ -17,7 +17,9 @@
 import random
 import re
 import shutil
+import stat
 import string
+import subprocess
 import sys
 import tempfile
 from datetime import datetime
@@ -1599,3 +1601,58 @@ def is_path_exists(uri: str, auth: Optional[Dict] = None) -> bool:
     if fsspec.filesystem(path_scheme, **storage_options).exists(uri):
         return True
     return False
+
+def apply_user_only_access_permissions(path: str):
+    """Applies user-only access permission to path. The logic is mainly taken reference from:
+    https://github.com/oracle/oci-cli/blob/master/src/oci_cli/cli_util.py#L2555
+
+    Parameters
+    ----------
+    path: str
+        Path to the file or folder
+    """
+    if not os.path.exists(path):
+        raise RuntimeError("Failed attempting to set permissions on path that does not exist: {}".format(path))
+
+    if is_windows():
+        # General permissions strategy is:
+        #   - if we create a new folder (e.g. C:\Users\opc\.oci), set access to allow full control for current user and no access for anyone else
+        #   - if we create a new file, set access to allow full control for current user and no access for anyone else
+        #   - thus if the user elects to place a new file (config or key) in an existing directory, we will not change the
+        #     permissions of that directory but will explicitly set the permissions on that file
+        username = os.environ['USERNAME']
+        userdomain = os.environ['UserDomain']
+        userWithDomain = os.environ['USERNAME']
+        if userdomain:
+            userWithDomain = userdomain + "\\" + username
+        admin_grp = '*S-1-5-32-544'
+        system_usr = '*S-1-5-18'
+        try:
+            if os.path.isfile(path):
+                subprocess.check_output('icacls "{path}" /reset'.format(path=path), stderr=subprocess.STDOUT)
+                try:
+                    subprocess.check_output('icacls "{path}" /inheritance:r /grant:r "{username}:F" /grant {admin_grp}:F /grant {system_usr}:F'.format(path=path, username=userWithDomain, admin_grp=admin_grp, system_usr=system_usr), stderr=subprocess.STDOUT)
+                except subprocess.CalledProcessError:
+                    subprocess.check_output('icacls "{path}" /inheritance:r /grant:r "{username}:F" /grant {admin_grp}:F /grant {system_usr}:F'.format(path=path, username=username, admin_grp=admin_grp, system_usr=system_usr), stderr=subprocess.STDOUT)
+            else:
+                if os.listdir(path):
+                    # safety check to make sure we aren't changing permissions of existing files
+                    raise RuntimeError("Failed attempting to set permissions on existing folder that is not empty.")
+                subprocess.check_output('icacls "{path}" /reset'.format(path=path), stderr=subprocess.STDOUT)
+                try:
+                    subprocess.check_output('icacls "{path}" /inheritance:r /grant:r "{username}:(OI)(CI)F"  /grant:r {admin_grp}:(OI)(CI)F /grant:r {system_usr}:(OI)(CI)F'.format(path=path, username=userWithDomain, admin_grp=admin_grp, system_usr=system_usr), stderr=subprocess.STDOUT)
+                except subprocess.CalledProcessError:
+                    subprocess.check_output('icacls "{path}" /inheritance:r /grant:r "{username}:(OI)(CI)F"  /grant:r {admin_grp}:(OI)(CI)F /grant:r {system_usr}:(OI)(CI)F'.format(path=path, username=username, admin_grp=admin_grp, system_usr=system_usr), stderr=subprocess.STDOUT)
+        except subprocess.CalledProcessError as exc_info:
+            print("Error occurred while attempting to set permissions for {path}: {exception}".format(path=path, exception=str(exc_info)))
+            sys.exit(exc_info.returncode)
+    else:
+        if os.path.isfile(path):
+            os.chmod(path, stat.S_IRUSR | stat.S_IWUSR)
+        else:
+            # For directories, we need to apply S_IXUSER otherwise it looks like on Linux/Unix/macOS if we create the directory then
+            # it won't behave like a directory and let files be put into it
+            os.chmod(path, stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)
+
+def is_windows():
+    return sys.platform == 'win32' or sys.platform == 'cygwin'

tests/unitary/default_setup/auth/test_auth.py

@@ -638,7 +638,7 @@ def test_validate_and_refresh_token(
         mock_get_jwt.assert_called()
         mock_refresh_security_token.assert_called_with(configuration)
 
-    @mock.patch("oci_cli.cli_util.apply_user_only_access_permissions")
+    @mock.patch("ads.common.utils.apply_user_only_access_permissions")
     @mock.patch("json.loads")
     @mock.patch("requests.post")
     @mock.patch("json.dumps")