Updated pr.

Author: lu-ohai

ads/common/auth.py

@@ -860,74 +860,24 @@ def _validate_and_refresh_token(self, configuration: Dict[str, Any]):
 
         if not security_token_container.valid():
             raise SecurityTokenError(
-                "Security token is invalid or has expired. Call `oci session authenticate` to generate new session."
+                "Security token has expired. Call `oci session authenticate` to generate new session."
             )
 
         time_now = int(time.time())
         time_expired = security_token_container.get_jwt()["exp"]
         if time_now - time_expired < SECURITY_TOKEN_LEFT_TIME:
-            try:
-                self._refresh_security_token(configuration)
-            except Exception as ex:
-                logger.info("Failed to refresh security token. Error: {}".format(ex))
+            if not self.oci_config_location:
+                logger.warning("Can not auto-refresh token. Specify parameter `oci_config_location` through ads.set_auth() or ads.auth.create_signer().")
+            else:
+                result = os.system(f"oci session refresh --config-file {self.oci_config_location} --profile {self.oci_key_profile}")
+                if result == 1:
+                    logger.warning(
+                        "Some error happened during auto-refreshing the token. Continue using the current one that's expiring in less than {SECURITY_TOKEN_LEFT_TIME} seconds."
+                        "Please follow steps in https://docs.oracle.com/en-us/iaas/Content/API/SDKDocs/clitoken.htm to renew token."
+                    )
 
         date_time = datetime.fromtimestamp(time_expired).strftime("%Y-%m-%d %H:%M:%S")
         logger.info(f"Session is valid until {date_time}.")
-    
-    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/master/src/oci_cli/cli_session.py#L152
-
-        Parameters
-        ----------
-        configuration: Dict
-            Security token configuration.
-        """
-        expanded_security_token_location = os.path.expanduser(
-            configuration.get("security_token_file")
-        )
-
-        with open(expanded_security_token_location, 'r') as security_token_file:
-            token = security_token_file.read()
-
-        try:
-            private_key = oci.signer.load_private_key_from_file(
-                configuration.get("key_file"), configuration.get("pass_phrase")
-            )
-        except:
-            raise
-        auth = oci.auth.signers.SecurityTokenSigner(token, private_key)
-
-        refresh_url = "{endpoint}/v1/authentication/refresh".format(
-            endpoint=oci.regions.endpoint_for("auth", configuration.get("region"))
-        )
-        logger.info(f"Attempting to refresh token from {refresh_url}.")
-
-        response = requests.post(
-            refresh_url,
-            headers={
-                'content-type': 'application/json'
-            },
-            data=json.dumps({
-                'currentToken': token
-            }),
-            auth=auth
-        )
-
-        if response.status_code == 200:
-            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)
-            utils.apply_user_only_access_permissions(expanded_security_token_location)
-            logger.info("Successfully refreshed token")
-        elif response.status_code == 401:
-            raise SecurityTokenError(
-                "Security token has expired. Call `oci session authenticate` to generate new session."
-            )
-        else:
-            raise SecurityTokenError(
-                "Failed to refresh sesison. Error: {}".format(str(response.content.decode('UTF-8')))
-            )
 
     def _read_security_token_file(self, security_token_file: str) -> str:
         """Reads security token from file.

ads/common/utils.py

@@ -17,9 +17,7 @@
 import random
 import re
 import shutil
-import stat
 import string
-import subprocess
 import sys
 import tempfile
 from datetime import datetime
@@ -1601,58 +1599,3 @@ 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

@@ -586,7 +586,7 @@ def test_security_token(
         assert signer["config"]["key_file"] == "test_key_file"
         assert isinstance(signer["signer"], SecurityTokenSigner)
 
-    @mock.patch("ads.common.auth.SecurityToken._refresh_security_token")
+    @mock.patch("os.system")
     @mock.patch("oci.auth.security_token_container.SecurityTokenContainer.get_jwt")
     @mock.patch("time.time")
     @mock.patch("oci.auth.security_token_container.SecurityTokenContainer.valid")
@@ -599,7 +599,7 @@ def test_validate_and_refresh_token(
         mock_valid,
         mock_time,
         mock_get_jwt,
-        mock_refresh_security_token
+        mock_system
     ):
         security_token = SecurityToken(
             args={
@@ -625,79 +625,18 @@ def test_validate_and_refresh_token(
         ):
             security_token._validate_and_refresh_token(configuration)
 
-        
         mock_valid.return_value = True
         mock_time.return_value = 1
         mock_get_jwt.return_value = {"exp" : 1}
+        mock_system.return_value = 1
 
         security_token._validate_and_refresh_token(configuration)
 
         mock_read_security_token_file.assert_called_with("test_security_token")
         mock_security_token_container.assert_called()
         mock_time.assert_called()
         mock_get_jwt.assert_called()
-        mock_refresh_security_token.assert_called_with(configuration)
-
-    @mock.patch("ads.common.utils.apply_user_only_access_permissions")
-    @mock.patch("json.loads")
-    @mock.patch("requests.post")
-    @mock.patch("json.dumps")
-    @mock.patch("oci.auth.signers.SecurityTokenSigner.__init__")
-    @mock.patch("oci.signer.load_private_key_from_file")
-    @mock.patch("builtins.open")
-    def test_refresh_security_token(
-        self,
-        mock_open,
-        mock_load_private_key_from_file,
-        mock_security_token_signer,
-        mock_dumps,
-        mock_post,
-        mock_loads,
-        mock_apply_user_only_access_permissions
-    ):
-        security_token = SecurityToken(args={})
-        configuration = {
-            "fingerprint": "test_fingerprint",
-            "tenancy": "test_tenancy",
-            "region": "us-ashburn-1",
-            "key_file": "test_key_file",
-            "security_token_file": "test_security_token",
-            "generic_headers": [1,2,3],
-            "body_headers": [4,5,6]
-        }
-        mock_security_token_signer.return_value = None
-        mock_loads.return_value = {
-            "token": "test_token"
-        }
-
-        response = MagicMock()
-        response.status_code = 401
-        mock_post.return_value = response
-        with pytest.raises(
-            SecurityTokenError,
-            match="Security token has expired. Call `oci session authenticate` to generate new session."
-        ):
-            security_token._refresh_security_token(configuration)
-
-        response.status_code = 500
-        mock_post.return_value = response
-        with pytest.raises(
-            SecurityTokenError,
-        ):
-            security_token._refresh_security_token(configuration)
-
-        response.status_code = 200
-        response.content = bytes("test_content", encoding='utf8')
-        mock_post.return_value = response
-        security_token._refresh_security_token(configuration)
-
-        mock_open.assert_called()
-        mock_load_private_key_from_file.assert_called_with("test_key_file", None)
-        mock_security_token_signer.assert_called()
-        mock_dumps.assert_called()
-        mock_post.assert_called()
-        mock_loads.assert_called()
-        mock_apply_user_only_access_permissions.assert_called()
+        mock_system.assert_called_with(f"oci session refresh --config-file {DEFAULT_LOCATION} --profile test_profile")
 
     @mock.patch("builtins.open")
     @mock.patch("os.path.isfile")