We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Azure anonymous access fails when secret is configured
/kind bug

**What steps did you take and what happened:**
Create the azure azcreds secret
Attempt to use a public blob
Access fails since the SP does not have access to the blob

**What did you expect to happen:**
KFServing should attempt anonymous access before using the secret

**Anything else you would like to add:**
[Miscellaneous information that will assist in solving the issue.]


**Environment:**

- Istio Version:
- Knative Version:
- KFServing Version:
- Kubeflow version:
- Minikube version:
- Kubernetes version: (use `kubectl version`):
- OS (e.g. from `/etc/os-release`):


--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `python/kfserving/kfserving/storage.py`
Content:
```
1 # Copyright 2019 kubeflow.org.
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #    http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 
15 import glob
16 import logging
17 import tempfile
18 import os
19 import re
20 from azure.storage.blob import BlockBlobService
21 from google.auth import exceptions
22 from google.cloud import storage
23 from minio import Minio
24 
25 _GCS_PREFIX = "gs://"
26 _S3_PREFIX = "s3://"
27 _BLOB_RE = "https://(.+?).blob.core.windows.net/(.+)"
28 _LOCAL_PREFIX = "file://"
29 
30 
31 class Storage(object): # pylint: disable=too-few-public-methods
32     @staticmethod
33     def download(uri: str, out_dir: str = None) -> str:
34         logging.info("Copying contents of %s to local", uri)
35 
36         is_local = False
37         if uri.startswith(_LOCAL_PREFIX) or os.path.exists(uri):
38             is_local = True
39 
40         if out_dir is None:
41             if is_local:
42                 # noop if out_dir is not set and the path is local
43                 return Storage._download_local(uri)
44             out_dir = tempfile.mkdtemp()
45 
46         if uri.startswith(_GCS_PREFIX):
47             Storage._download_gcs(uri, out_dir)
48         elif uri.startswith(_S3_PREFIX):
49             Storage._download_s3(uri, out_dir)
50         elif re.search(_BLOB_RE, uri):
51             Storage._download_blob(uri, out_dir)
52         elif is_local:
53             return Storage._download_local(uri, out_dir)
54         else:
55             raise Exception("Cannot recognize storage type for " + uri +
56                             "\n'%s', '%s', and '%s' are the current available storage type." %
57                             (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX))
58 
59         logging.info("Successfully copied %s to %s", uri, out_dir)
60         return out_dir
61 
62     @staticmethod
63     def _download_s3(uri, temp_dir: str):
64         client = Storage._create_minio_client()
65         bucket_args = uri.replace(_S3_PREFIX, "", 1).split("/", 1)
66         bucket_name = bucket_args[0]
67         bucket_path = bucket_args[1] if len(bucket_args) > 1 else ""
68         objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)
69         for obj in objects:
70             # Replace any prefix from the object key with temp_dir
71             subdir_object_key = obj.object_name.replace(bucket_path, "", 1).strip("/")
72             client.fget_object(bucket_name, obj.object_name,
73                                os.path.join(temp_dir, subdir_object_key))
74 
75     @staticmethod
76     def _download_gcs(uri, temp_dir: str):
77         try:
78             storage_client = storage.Client()
79         except exceptions.DefaultCredentialsError:
80             storage_client = storage.Client.create_anonymous_client()
81         bucket_args = uri.replace(_GCS_PREFIX, "", 1).split("/", 1)
82         bucket_name = bucket_args[0]
83         bucket_path = bucket_args[1] if len(bucket_args) > 1 else ""
84         bucket = storage_client.bucket(bucket_name)
85         prefix = bucket_path
86         if not prefix.endswith("/"):
87             prefix = prefix + "/"
88         blobs = bucket.list_blobs(prefix=prefix)
89         for blob in blobs:
90             # Replace any prefix from the object key with temp_dir
91             subdir_object_key = blob.name.replace(bucket_path, "", 1).strip("/")
92 
93             # Create necessary subdirectory to store the object locally
94             if "/" in subdir_object_key:
95                 local_object_dir = os.path.join(temp_dir, subdir_object_key.rsplit("/", 1)[0])
96                 if not os.path.isdir(local_object_dir):
97                     os.makedirs(local_object_dir, exist_ok=True)
98             if subdir_object_key.strip() != "":
99                 dest_path = os.path.join(temp_dir, subdir_object_key)
100                 logging.info("Downloading: %s", dest_path)
101                 blob.download_to_filename(dest_path)
102 
103     @staticmethod
104     def _download_blob(uri, out_dir: str): # pylint: disable=too-many-locals
105         match = re.search(_BLOB_RE, uri)
106         account_name = match.group(1)
107         storage_url = match.group(2)
108         container_name, prefix = storage_url.split("/", 1)
109 
110         logging.info("Connecting to BLOB account: [%s], container: [%s], prefix: [%s]",
111                      account_name,
112                      container_name,
113                      prefix)
114         token = Storage._get_azure_storage_token()
115         block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)
116         blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
117 
118         for blob in blobs:
119             dest_path = os.path.join(out_dir, blob.name)
120             if "/" in blob.name:
121                 head, tail = os.path.split(blob.name)
122                 if prefix is not None:
123                     head = head[len(prefix):]
124                 if head.startswith('/'):
125                     head = head[1:]
126                 dir_path = os.path.join(out_dir, head)
127                 dest_path = os.path.join(dir_path, tail)
128                 if not os.path.isdir(dir_path):
129                     os.makedirs(dir_path)
130 
131             logging.info("Downloading: %s to %s", blob.name, dest_path)
132             block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)
133 
134     @staticmethod
135     def _get_azure_storage_token():
136         tenant_id = os.getenv("AZ_TENANT_ID", "")
137         client_id = os.getenv("AZ_CLIENT_ID", "")
138         client_secret = os.getenv("AZ_CLIENT_SECRET", "")
139         subscription_id = os.getenv("AZ_SUBSCRIPTION_ID", "")
140 
141         if tenant_id == "" or client_id == "" or client_secret == "" or subscription_id == "":
142             return None
143 
144         # note the SP must have "Storage Blob Data Owner" perms for this to work
145         import adal
146         from azure.storage.common import TokenCredential
147 
148         authority_url = "https://login.microsoftonline.com/" + tenant_id
149 
150         context = adal.AuthenticationContext(authority_url)
151 
152         token = context.acquire_token_with_client_credentials(
153             "https://storage.azure.com/",
154             client_id,
155             client_secret)
156 
157         token_credential = TokenCredential(token["accessToken"])
158 
159         return token_credential
160 
161     @staticmethod
162     def _download_local(uri, out_dir=None):
163         local_path = uri.replace(_LOCAL_PREFIX, "", 1)
164         if not os.path.exists(local_path):
165             raise Exception("Local path %s does not exist." % (uri))
166 
167         if out_dir is None:
168             return local_path
169         elif not os.path.isdir(out_dir):
170             os.makedirs(out_dir)
171 
172         if os.path.isdir(local_path):
173             local_path = os.path.join(local_path, "*")
174 
175         for src in glob.glob(local_path):
176             _, tail = os.path.split(src)
177             dest_path = os.path.join(out_dir, tail)
178             logging.info("Linking: %s to %s", src, dest_path)
179             os.symlink(src, dest_path)
180         return out_dir
181 
182     @staticmethod
183     def _create_minio_client():
184         # Remove possible http scheme for Minio
185         url = re.compile(r"https?://")
186         minioClient = Minio(url.sub("", os.getenv("S3_ENDPOINT", "")),
187                             access_key=os.getenv("AWS_ACCESS_KEY_ID", ""),
188                             secret_key=os.getenv("AWS_SECRET_ACCESS_KEY", ""),
189                             secure=True)
190         return minioClient
191
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/python/kfserving/kfserving/storage.py b/python/kfserving/kfserving/storage.py
--- a/python/kfserving/kfserving/storage.py
+++ b/python/kfserving/kfserving/storage.py
@@ -17,6 +17,7 @@
 import tempfile
 import os
 import re
+from azure.common import AzureMissingResourceHttpError
 from azure.storage.blob import BlockBlobService
 from google.auth import exceptions
 from google.cloud import storage
@@ -111,9 +112,15 @@
                      account_name,
                      container_name,
                      prefix)
-        token = Storage._get_azure_storage_token()
-        block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)
-        blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
+        try:
+            block_blob_service = BlockBlobService(account_name=account_name)
+            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
+        except AzureMissingResourceHttpError:
+            token = Storage._get_azure_storage_token()
+            if token is None:
+                logging.warning("Azure credentials not found, retrying anonymous access")
+            block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)
+            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
 
         for blob in blobs:
             dest_path = os.path.join(out_dir, blob.name)
@@ -156,6 +163,8 @@
 
         token_credential = TokenCredential(token["accessToken"])
 
+        logging.info("Retrieved SP token credential for client_id: %s", client_id)
+
         return token_credential
 
     @staticmethod

{"golden_diff": "diff --git a/python/kfserving/kfserving/storage.py b/python/kfserving/kfserving/storage.py\n--- a/python/kfserving/kfserving/storage.py\n+++ b/python/kfserving/kfserving/storage.py\n@@ -17,6 +17,7 @@\n import tempfile\n import os\n import re\n+from azure.common import AzureMissingResourceHttpError\n from azure.storage.blob import BlockBlobService\n from google.auth import exceptions\n from google.cloud import storage\n@@ -111,9 +112,15 @@\n                      account_name,\n                      container_name,\n                      prefix)\n-        token = Storage._get_azure_storage_token()\n-        block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)\n-        blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n+        try:\n+            block_blob_service = BlockBlobService(account_name=account_name)\n+            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n+        except AzureMissingResourceHttpError:\n+            token = Storage._get_azure_storage_token()\n+            if token is None:\n+                logging.warning(\"Azure credentials not found, retrying anonymous access\")\n+            block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)\n+            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n \n         for blob in blobs:\n             dest_path = os.path.join(out_dir, blob.name)\n@@ -156,6 +163,8 @@\n \n         token_credential = TokenCredential(token[\"accessToken\"])\n \n+        logging.info(\"Retrieved SP token credential for client_id: %s\", client_id)\n+\n         return token_credential\n \n     @staticmethod\n", "issue": "Azure anonymous access fails when secret is configured\n/kind bug\r\n\r\n**What steps did you take and what happened:**\r\nCreate the azure azcreds secret\r\nAttempt to use a public blob\r\nAccess fails since the SP does not have access to the blob\r\n\r\n**What did you expect to happen:**\r\nKFServing should attempt anonymous access before using the secret\r\n\r\n**Anything else you would like to add:**\r\n[Miscellaneous information that will assist in solving the issue.]\r\n\r\n\r\n**Environment:**\r\n\r\n- Istio Version:\r\n- Knative Version:\r\n- KFServing Version:\r\n- Kubeflow version:\r\n- Minikube version:\r\n- Kubernetes version: (use `kubectl version`):\r\n- OS (e.g. from `/etc/os-release`):\r\n\n", "before_files": [{"content": "# Copyright 2019 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport glob\nimport logging\nimport tempfile\nimport os\nimport re\nfrom azure.storage.blob import BlockBlobService\nfrom google.auth import exceptions\nfrom google.cloud import storage\nfrom minio import Minio\n\n_GCS_PREFIX = \"gs://\"\n_S3_PREFIX = \"s3://\"\n_BLOB_RE = \"https://(.+?).blob.core.windows.net/(.+)\"\n_LOCAL_PREFIX = \"file://\"\n\n\nclass Storage(object): # pylint: disable=too-few-public-methods\n    @staticmethod\n    def download(uri: str, out_dir: str = None) -> str:\n        logging.info(\"Copying contents of %s to local\", uri)\n\n        is_local = False\n        if uri.startswith(_LOCAL_PREFIX) or os.path.exists(uri):\n            is_local = True\n\n        if out_dir is None:\n            if is_local:\n                # noop if out_dir is not set and the path is local\n                return Storage._download_local(uri)\n            out_dir = tempfile.mkdtemp()\n\n        if uri.startswith(_GCS_PREFIX):\n            Storage._download_gcs(uri, out_dir)\n        elif uri.startswith(_S3_PREFIX):\n            Storage._download_s3(uri, out_dir)\n        elif re.search(_BLOB_RE, uri):\n            Storage._download_blob(uri, out_dir)\n        elif is_local:\n            return Storage._download_local(uri, out_dir)\n        else:\n            raise Exception(\"Cannot recognize storage type for \" + uri +\n                            \"\\n'%s', '%s', and '%s' are the current available storage type.\" %\n                            (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX))\n\n        logging.info(\"Successfully copied %s to %s\", uri, out_dir)\n        return out_dir\n\n    @staticmethod\n    def _download_s3(uri, temp_dir: str):\n        client = Storage._create_minio_client()\n        bucket_args = uri.replace(_S3_PREFIX, \"\", 1).split(\"/\", 1)\n        bucket_name = bucket_args[0]\n        bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n        objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)\n        for obj in objects:\n            # Replace any prefix from the object key with temp_dir\n            subdir_object_key = obj.object_name.replace(bucket_path, \"\", 1).strip(\"/\")\n            client.fget_object(bucket_name, obj.object_name,\n                               os.path.join(temp_dir, subdir_object_key))\n\n    @staticmethod\n    def _download_gcs(uri, temp_dir: str):\n        try:\n            storage_client = storage.Client()\n        except exceptions.DefaultCredentialsError:\n            storage_client = storage.Client.create_anonymous_client()\n        bucket_args = uri.replace(_GCS_PREFIX, \"\", 1).split(\"/\", 1)\n        bucket_name = bucket_args[0]\n        bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n        bucket = storage_client.bucket(bucket_name)\n        prefix = bucket_path\n        if not prefix.endswith(\"/\"):\n            prefix = prefix + \"/\"\n        blobs = bucket.list_blobs(prefix=prefix)\n        for blob in blobs:\n            # Replace any prefix from the object key with temp_dir\n            subdir_object_key = blob.name.replace(bucket_path, \"\", 1).strip(\"/\")\n\n            # Create necessary subdirectory to store the object locally\n            if \"/\" in subdir_object_key:\n                local_object_dir = os.path.join(temp_dir, subdir_object_key.rsplit(\"/\", 1)[0])\n                if not os.path.isdir(local_object_dir):\n                    os.makedirs(local_object_dir, exist_ok=True)\n            if subdir_object_key.strip() != \"\":\n                dest_path = os.path.join(temp_dir, subdir_object_key)\n                logging.info(\"Downloading: %s\", dest_path)\n                blob.download_to_filename(dest_path)\n\n    @staticmethod\n    def _download_blob(uri, out_dir: str): # pylint: disable=too-many-locals\n        match = re.search(_BLOB_RE, uri)\n        account_name = match.group(1)\n        storage_url = match.group(2)\n        container_name, prefix = storage_url.split(\"/\", 1)\n\n        logging.info(\"Connecting to BLOB account: [%s], container: [%s], prefix: [%s]\",\n                     account_name,\n                     container_name,\n                     prefix)\n        token = Storage._get_azure_storage_token()\n        block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)\n        blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n\n        for blob in blobs:\n            dest_path = os.path.join(out_dir, blob.name)\n            if \"/\" in blob.name:\n                head, tail = os.path.split(blob.name)\n                if prefix is not None:\n                    head = head[len(prefix):]\n                if head.startswith('/'):\n                    head = head[1:]\n                dir_path = os.path.join(out_dir, head)\n                dest_path = os.path.join(dir_path, tail)\n                if not os.path.isdir(dir_path):\n                    os.makedirs(dir_path)\n\n            logging.info(\"Downloading: %s to %s\", blob.name, dest_path)\n            block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)\n\n    @staticmethod\n    def _get_azure_storage_token():\n        tenant_id = os.getenv(\"AZ_TENANT_ID\", \"\")\n        client_id = os.getenv(\"AZ_CLIENT_ID\", \"\")\n        client_secret = os.getenv(\"AZ_CLIENT_SECRET\", \"\")\n        subscription_id = os.getenv(\"AZ_SUBSCRIPTION_ID\", \"\")\n\n        if tenant_id == \"\" or client_id == \"\" or client_secret == \"\" or subscription_id == \"\":\n            return None\n\n        # note the SP must have \"Storage Blob Data Owner\" perms for this to work\n        import adal\n        from azure.storage.common import TokenCredential\n\n        authority_url = \"https://login.microsoftonline.com/\" + tenant_id\n\n        context = adal.AuthenticationContext(authority_url)\n\n        token = context.acquire_token_with_client_credentials(\n            \"https://storage.azure.com/\",\n            client_id,\n            client_secret)\n\n        token_credential = TokenCredential(token[\"accessToken\"])\n\n        return token_credential\n\n    @staticmethod\n    def _download_local(uri, out_dir=None):\n        local_path = uri.replace(_LOCAL_PREFIX, \"\", 1)\n        if not os.path.exists(local_path):\n            raise Exception(\"Local path %s does not exist.\" % (uri))\n\n        if out_dir is None:\n            return local_path\n        elif not os.path.isdir(out_dir):\n            os.makedirs(out_dir)\n\n        if os.path.isdir(local_path):\n            local_path = os.path.join(local_path, \"*\")\n\n        for src in glob.glob(local_path):\n            _, tail = os.path.split(src)\n            dest_path = os.path.join(out_dir, tail)\n            logging.info(\"Linking: %s to %s\", src, dest_path)\n            os.symlink(src, dest_path)\n        return out_dir\n\n    @staticmethod\n    def _create_minio_client():\n        # Remove possible http scheme for Minio\n        url = re.compile(r\"https?://\")\n        minioClient = Minio(url.sub(\"\", os.getenv(\"S3_ENDPOINT\", \"\")),\n                            access_key=os.getenv(\"AWS_ACCESS_KEY_ID\", \"\"),\n                            secret_key=os.getenv(\"AWS_SECRET_ACCESS_KEY\", \"\"),\n                            secure=True)\n        return minioClient\n", "path": "python/kfserving/kfserving/storage.py"}], "after_files": [{"content": "# Copyright 2019 kubeflow.org.\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#    http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\nimport glob\nimport logging\nimport tempfile\nimport os\nimport re\nfrom azure.common import AzureMissingResourceHttpError\nfrom azure.storage.blob import BlockBlobService\nfrom google.auth import exceptions\nfrom google.cloud import storage\nfrom minio import Minio\n\n_GCS_PREFIX = \"gs://\"\n_S3_PREFIX = \"s3://\"\n_BLOB_RE = \"https://(.+?).blob.core.windows.net/(.+)\"\n_LOCAL_PREFIX = \"file://\"\n\n\nclass Storage(object): # pylint: disable=too-few-public-methods\n    @staticmethod\n    def download(uri: str, out_dir: str = None) -> str:\n        logging.info(\"Copying contents of %s to local\", uri)\n\n        is_local = False\n        if uri.startswith(_LOCAL_PREFIX) or os.path.exists(uri):\n            is_local = True\n\n        if out_dir is None:\n            if is_local:\n                # noop if out_dir is not set and the path is local\n                return Storage._download_local(uri)\n            out_dir = tempfile.mkdtemp()\n\n        if uri.startswith(_GCS_PREFIX):\n            Storage._download_gcs(uri, out_dir)\n        elif uri.startswith(_S3_PREFIX):\n            Storage._download_s3(uri, out_dir)\n        elif re.search(_BLOB_RE, uri):\n            Storage._download_blob(uri, out_dir)\n        elif is_local:\n            return Storage._download_local(uri, out_dir)\n        else:\n            raise Exception(\"Cannot recognize storage type for \" + uri +\n                            \"\\n'%s', '%s', and '%s' are the current available storage type.\" %\n                            (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX))\n\n        logging.info(\"Successfully copied %s to %s\", uri, out_dir)\n        return out_dir\n\n    @staticmethod\n    def _download_s3(uri, temp_dir: str):\n        client = Storage._create_minio_client()\n        bucket_args = uri.replace(_S3_PREFIX, \"\", 1).split(\"/\", 1)\n        bucket_name = bucket_args[0]\n        bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n        objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)\n        for obj in objects:\n            # Replace any prefix from the object key with temp_dir\n            subdir_object_key = obj.object_name.replace(bucket_path, \"\", 1).strip(\"/\")\n            client.fget_object(bucket_name, obj.object_name,\n                               os.path.join(temp_dir, subdir_object_key))\n\n    @staticmethod\n    def _download_gcs(uri, temp_dir: str):\n        try:\n            storage_client = storage.Client()\n        except exceptions.DefaultCredentialsError:\n            storage_client = storage.Client.create_anonymous_client()\n        bucket_args = uri.replace(_GCS_PREFIX, \"\", 1).split(\"/\", 1)\n        bucket_name = bucket_args[0]\n        bucket_path = bucket_args[1] if len(bucket_args) > 1 else \"\"\n        bucket = storage_client.bucket(bucket_name)\n        prefix = bucket_path\n        if not prefix.endswith(\"/\"):\n            prefix = prefix + \"/\"\n        blobs = bucket.list_blobs(prefix=prefix)\n        for blob in blobs:\n            # Replace any prefix from the object key with temp_dir\n            subdir_object_key = blob.name.replace(bucket_path, \"\", 1).strip(\"/\")\n\n            # Create necessary subdirectory to store the object locally\n            if \"/\" in subdir_object_key:\n                local_object_dir = os.path.join(temp_dir, subdir_object_key.rsplit(\"/\", 1)[0])\n                if not os.path.isdir(local_object_dir):\n                    os.makedirs(local_object_dir, exist_ok=True)\n            if subdir_object_key.strip() != \"\":\n                dest_path = os.path.join(temp_dir, subdir_object_key)\n                logging.info(\"Downloading: %s\", dest_path)\n                blob.download_to_filename(dest_path)\n\n    @staticmethod\n    def _download_blob(uri, out_dir: str): # pylint: disable=too-many-locals\n        match = re.search(_BLOB_RE, uri)\n        account_name = match.group(1)\n        storage_url = match.group(2)\n        container_name, prefix = storage_url.split(\"/\", 1)\n\n        logging.info(\"Connecting to BLOB account: [%s], container: [%s], prefix: [%s]\",\n                     account_name,\n                     container_name,\n                     prefix)\n        try:\n            block_blob_service = BlockBlobService(account_name=account_name)\n            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n        except AzureMissingResourceHttpError:\n            token = Storage._get_azure_storage_token()\n            if token is None:\n                logging.warning(\"Azure credentials not found, retrying anonymous access\")\n            block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)\n            blobs = block_blob_service.list_blobs(container_name, prefix=prefix)\n\n        for blob in blobs:\n            dest_path = os.path.join(out_dir, blob.name)\n            if \"/\" in blob.name:\n                head, tail = os.path.split(blob.name)\n                if prefix is not None:\n                    head = head[len(prefix):]\n                if head.startswith('/'):\n                    head = head[1:]\n                dir_path = os.path.join(out_dir, head)\n                dest_path = os.path.join(dir_path, tail)\n                if not os.path.isdir(dir_path):\n                    os.makedirs(dir_path)\n\n            logging.info(\"Downloading: %s to %s\", blob.name, dest_path)\n            block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)\n\n    @staticmethod\n    def _get_azure_storage_token():\n        tenant_id = os.getenv(\"AZ_TENANT_ID\", \"\")\n        client_id = os.getenv(\"AZ_CLIENT_ID\", \"\")\n        client_secret = os.getenv(\"AZ_CLIENT_SECRET\", \"\")\n        subscription_id = os.getenv(\"AZ_SUBSCRIPTION_ID\", \"\")\n\n        if tenant_id == \"\" or client_id == \"\" or client_secret == \"\" or subscription_id == \"\":\n            return None\n\n        # note the SP must have \"Storage Blob Data Owner\" perms for this to work\n        import adal\n        from azure.storage.common import TokenCredential\n\n        authority_url = \"https://login.microsoftonline.com/\" + tenant_id\n\n        context = adal.AuthenticationContext(authority_url)\n\n        token = context.acquire_token_with_client_credentials(\n            \"https://storage.azure.com/\",\n            client_id,\n            client_secret)\n\n        token_credential = TokenCredential(token[\"accessToken\"])\n\n        logging.info(\"Retrieved SP token credential for client_id: %s\", client_id)\n\n        return token_credential\n\n    @staticmethod\n    def _download_local(uri, out_dir=None):\n        local_path = uri.replace(_LOCAL_PREFIX, \"\", 1)\n        if not os.path.exists(local_path):\n            raise Exception(\"Local path %s does not exist.\" % (uri))\n\n        if out_dir is None:\n            return local_path\n        elif not os.path.isdir(out_dir):\n            os.makedirs(out_dir)\n\n        if os.path.isdir(local_path):\n            local_path = os.path.join(local_path, \"*\")\n\n        for src in glob.glob(local_path):\n            _, tail = os.path.split(src)\n            dest_path = os.path.join(out_dir, tail)\n            logging.info(\"Linking: %s to %s\", src, dest_path)\n            os.symlink(src, dest_path)\n        return out_dir\n\n    @staticmethod\n    def _create_minio_client():\n        # Remove possible http scheme for Minio\n        url = re.compile(r\"https?://\")\n        minioClient = Minio(url.sub(\"\", os.getenv(\"S3_ENDPOINT\", \"\")),\n                            access_key=os.getenv(\"AWS_ACCESS_KEY_ID\", \"\"),\n                            secret_key=os.getenv(\"AWS_SECRET_ACCESS_KEY\", \"\"),\n                            secure=True)\n        return minioClient\n", "path": "python/kfserving/kfserving/storage.py"}]}