We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
[AWS] Support IAM Role for Service Account in KFServing
/kind feature

**Describe the solution you'd like**
[A clear and concise description of what you want to happen.]
Currently, it needs a S3 credential to download model. We need more fine grain control ways. [IRSA](https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html) is a better option. This helps us deploy model without credentials
 
**Anything else you would like to add:**
[Miscellaneous information that will assist in solving the issue.]

--- 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 2020 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 mimetypes
19 import os
20 import re
21 import json
22 import shutil
23 import tarfile
24 import zipfile
25 import gzip
26 from urllib.parse import urlparse
27 import requests
28 from azure.storage.blob import BlockBlobService
29 from google.auth import exceptions
30 from google.cloud import storage
31 from minio import Minio
32 from kfserving.kfmodel_repository import MODEL_MOUNT_DIRS
33 
34 _GCS_PREFIX = "gs://"
35 _S3_PREFIX = "s3://"
36 _BLOB_RE = "https://(.+?).blob.core.windows.net/(.+)"
37 _LOCAL_PREFIX = "file://"
38 _URI_RE = "https?://(.+)/(.+)"
39 _HTTP_PREFIX = "http(s)://"
40 _HEADERS_SUFFIX = "-headers"
41 
42 
43 class Storage(object):  # pylint: disable=too-few-public-methods
44     @staticmethod
45     def download(uri: str, out_dir: str = None) -> str:
46         logging.info("Copying contents of %s to local", uri)
47 
48         is_local = False
49         if uri.startswith(_LOCAL_PREFIX) or os.path.exists(uri):
50             is_local = True
51 
52         if out_dir is None:
53             if is_local:
54                 # noop if out_dir is not set and the path is local
55                 return Storage._download_local(uri)
56             out_dir = tempfile.mkdtemp()
57         elif not os.path.exists(out_dir):
58             os.mkdir(out_dir)
59 
60         if uri.startswith(_GCS_PREFIX):
61             Storage._download_gcs(uri, out_dir)
62         elif uri.startswith(_S3_PREFIX):
63             Storage._download_s3(uri, out_dir)
64         elif re.search(_BLOB_RE, uri):
65             Storage._download_blob(uri, out_dir)
66         elif is_local:
67             return Storage._download_local(uri, out_dir)
68         elif re.search(_URI_RE, uri):
69             return Storage._download_from_uri(uri, out_dir)
70         elif uri.startswith(MODEL_MOUNT_DIRS):
71             # Don't need to download models if this InferenceService is running in the multi-model
72             # serving mode. The model agent will download models.
73             return out_dir
74         else:
75             raise Exception("Cannot recognize storage type for " + uri +
76                             "\n'%s', '%s', '%s', and '%s' are the current available storage type." %
77                             (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX, _HTTP_PREFIX))
78 
79         logging.info("Successfully copied %s to %s", uri, out_dir)
80         return out_dir
81 
82     @staticmethod
83     def _download_s3(uri, temp_dir: str):
84         client = Storage._create_minio_client()
85         bucket_args = uri.replace(_S3_PREFIX, "", 1).split("/", 1)
86         bucket_name = bucket_args[0]
87         bucket_path = bucket_args[1] if len(bucket_args) > 1 else ""
88         objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)
89         count = 0
90         for obj in objects:
91             # Replace any prefix from the object key with temp_dir
92             subdir_object_key = obj.object_name.replace(bucket_path, "", 1).strip("/")
93             # fget_object handles directory creation if does not exist
94             if not obj.is_dir:
95                 if subdir_object_key == "":
96                     subdir_object_key = obj.object_name
97                 client.fget_object(bucket_name, obj.object_name,
98                                    os.path.join(temp_dir, subdir_object_key))
99             count = count + 1
100         if count == 0:
101             raise RuntimeError("Failed to fetch model. \
102 The path or model %s does not exist." % (uri))
103 
104     @staticmethod
105     def _download_gcs(uri, temp_dir: str):
106         try:
107             storage_client = storage.Client()
108         except exceptions.DefaultCredentialsError:
109             storage_client = storage.Client.create_anonymous_client()
110         bucket_args = uri.replace(_GCS_PREFIX, "", 1).split("/", 1)
111         bucket_name = bucket_args[0]
112         bucket_path = bucket_args[1] if len(bucket_args) > 1 else ""
113         bucket = storage_client.bucket(bucket_name)
114         prefix = bucket_path
115         if not prefix.endswith("/"):
116             prefix = prefix + "/"
117         blobs = bucket.list_blobs(prefix=prefix)
118         count = 0
119         for blob in blobs:
120             # Replace any prefix from the object key with temp_dir
121             subdir_object_key = blob.name.replace(bucket_path, "", 1).strip("/")
122 
123             # Create necessary subdirectory to store the object locally
124             if "/" in subdir_object_key:
125                 local_object_dir = os.path.join(temp_dir, subdir_object_key.rsplit("/", 1)[0])
126                 if not os.path.isdir(local_object_dir):
127                     os.makedirs(local_object_dir, exist_ok=True)
128             if subdir_object_key.strip() != "":
129                 dest_path = os.path.join(temp_dir, subdir_object_key)
130                 logging.info("Downloading: %s", dest_path)
131                 blob.download_to_filename(dest_path)
132             count = count + 1
133         if count == 0:
134             raise RuntimeError("Failed to fetch model. \
135 The path or model %s does not exist." % uri)
136 
137     @staticmethod
138     def _download_blob(uri, out_dir: str):  # pylint: disable=too-many-locals
139         match = re.search(_BLOB_RE, uri)
140         account_name = match.group(1)
141         storage_url = match.group(2)
142         container_name, prefix = storage_url.split("/", 1)
143 
144         logging.info("Connecting to BLOB account: [%s], container: [%s], prefix: [%s]",
145                      account_name,
146                      container_name,
147                      prefix)
148         try:
149             block_blob_service = BlockBlobService(account_name=account_name)
150             blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
151         except Exception:  # pylint: disable=broad-except
152             token = Storage._get_azure_storage_token()
153             if token is None:
154                 logging.warning("Azure credentials not found, retrying anonymous access")
155             block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)
156             blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
157         count = 0
158         for blob in blobs:
159             dest_path = os.path.join(out_dir, blob.name)
160             if "/" in blob.name:
161                 head, tail = os.path.split(blob.name)
162                 if prefix is not None:
163                     head = head[len(prefix):]
164                 if head.startswith('/'):
165                     head = head[1:]
166                 dir_path = os.path.join(out_dir, head)
167                 dest_path = os.path.join(dir_path, tail)
168                 if not os.path.isdir(dir_path):
169                     os.makedirs(dir_path)
170 
171             logging.info("Downloading: %s to %s", blob.name, dest_path)
172             block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)
173             count = count + 1
174         if count == 0:
175             raise RuntimeError("Failed to fetch model. \
176 The path or model %s does not exist." % (uri))
177 
178     @staticmethod
179     def _get_azure_storage_token():
180         tenant_id = os.getenv("AZ_TENANT_ID", "")
181         client_id = os.getenv("AZ_CLIENT_ID", "")
182         client_secret = os.getenv("AZ_CLIENT_SECRET", "")
183         subscription_id = os.getenv("AZ_SUBSCRIPTION_ID", "")
184 
185         if tenant_id == "" or client_id == "" or client_secret == "" or subscription_id == "":
186             return None
187 
188         # note the SP must have "Storage Blob Data Owner" perms for this to work
189         import adal
190         from azure.storage.common import TokenCredential
191 
192         authority_url = "https://login.microsoftonline.com/" + tenant_id
193 
194         context = adal.AuthenticationContext(authority_url)
195 
196         token = context.acquire_token_with_client_credentials(
197             "https://storage.azure.com/",
198             client_id,
199             client_secret)
200 
201         token_credential = TokenCredential(token["accessToken"])
202 
203         logging.info("Retrieved SP token credential for client_id: %s", client_id)
204 
205         return token_credential
206 
207     @staticmethod
208     def _download_local(uri, out_dir=None):
209         local_path = uri.replace(_LOCAL_PREFIX, "", 1)
210         if not os.path.exists(local_path):
211             raise RuntimeError("Local path %s does not exist." % (uri))
212 
213         if out_dir is None:
214             return local_path
215         elif not os.path.isdir(out_dir):
216             os.makedirs(out_dir)
217 
218         if os.path.isdir(local_path):
219             local_path = os.path.join(local_path, "*")
220 
221         for src in glob.glob(local_path):
222             _, tail = os.path.split(src)
223             dest_path = os.path.join(out_dir, tail)
224             logging.info("Linking: %s to %s", src, dest_path)
225             os.symlink(src, dest_path)
226         return out_dir
227 
228     @staticmethod
229     def _download_from_uri(uri, out_dir=None):
230         url = urlparse(uri)
231         filename = os.path.basename(url.path)
232         mimetype, encoding = mimetypes.guess_type(url.path)
233         local_path = os.path.join(out_dir, filename)
234 
235         if filename == '':
236             raise ValueError('No filename contained in URI: %s' % (uri))
237 
238         # Get header information from host url
239         headers = {}
240         host_uri = url.hostname
241 
242         headers_json = os.getenv(host_uri + _HEADERS_SUFFIX, "{}")
243         headers = json.loads(headers_json)
244 
245         with requests.get(uri, stream=True, headers=headers) as response:
246             if response.status_code != 200:
247                 raise RuntimeError("URI: %s returned a %s response code." % (uri, response.status_code))
248             if mimetype == 'application/zip' and not response.headers.get('Content-Type', '')\
249                     .startswith('application/zip'):
250                 raise RuntimeError("URI: %s did not respond with \'Content-Type\': \'application/zip\'" % uri)
251             if mimetype == 'application/x-tar' and not response.headers.get('Content-Type', '')\
252                     .startswith('application/x-tar'):
253                 raise RuntimeError("URI: %s did not respond with \'Content-Type\': \'application/x-tar\'" % uri)
254             if (mimetype != 'application/zip' and mimetype != 'application/x-tar') and \
255                     not response.headers.get('Content-Type', '').startswith('application/octet-stream'):
256                 raise RuntimeError("URI: %s did not respond with \'Content-Type\': \'application/octet-stream\'"
257                                    % uri)
258 
259             if encoding == 'gzip':
260                 stream = gzip.GzipFile(fileobj=response.raw)
261                 local_path = os.path.join(out_dir, f'{filename}.tar')
262             else:
263                 stream = response.raw
264             with open(local_path, 'wb') as out:
265                 shutil.copyfileobj(stream, out)
266 
267         if mimetype in ["application/x-tar", "application/zip"]:
268             if mimetype == "application/x-tar":
269                 archive = tarfile.open(local_path, 'r', encoding='utf-8')
270             else:
271                 archive = zipfile.ZipFile(local_path, 'r')
272             archive.extractall(out_dir)
273             archive.close()
274             os.remove(local_path)
275 
276         return out_dir
277 
278     @staticmethod
279     def _create_minio_client():
280         # Adding prefixing "http" in urlparse is necessary for it to be the netloc
281         url = urlparse(os.getenv("AWS_ENDPOINT_URL", "http://s3.amazonaws.com"))
282         use_ssl = url.scheme == 'https' if url.scheme else bool(os.getenv("S3_USE_HTTPS", "true"))
283         return Minio(url.netloc,
284                      access_key=os.getenv("AWS_ACCESS_KEY_ID", ""),
285                      secret_key=os.getenv("AWS_SECRET_ACCESS_KEY", ""),
286                      region=os.getenv("AWS_REGION", ""),
287                      secure=use_ssl)
288
```
--- 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
@@ -13,22 +13,24 @@
 # limitations under the License.
 
 import glob
+import gzip
 import logging
-import tempfile
 import mimetypes
 import os
 import re
 import json
 import shutil
 import tarfile
+import tempfile
 import zipfile
-import gzip
 from urllib.parse import urlparse
+
+import boto3
 import requests
 from azure.storage.blob import BlockBlobService
 from google.auth import exceptions
 from google.cloud import storage
-from minio import Minio
+
 from kfserving.kfmodel_repository import MODEL_MOUNT_DIRS
 
 _GCS_PREFIX = "gs://"
@@ -81,25 +83,27 @@
 
     @staticmethod
     def _download_s3(uri, temp_dir: str):
-        client = Storage._create_minio_client()
-        bucket_args = uri.replace(_S3_PREFIX, "", 1).split("/", 1)
-        bucket_name = bucket_args[0]
-        bucket_path = bucket_args[1] if len(bucket_args) > 1 else ""
-        objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)
-        count = 0
-        for obj in objects:
-            # Replace any prefix from the object key with temp_dir
-            subdir_object_key = obj.object_name.replace(bucket_path, "", 1).strip("/")
-            # fget_object handles directory creation if does not exist
-            if not obj.is_dir:
-                if subdir_object_key == "":
-                    subdir_object_key = obj.object_name
-                client.fget_object(bucket_name, obj.object_name,
-                                   os.path.join(temp_dir, subdir_object_key))
-            count = count + 1
-        if count == 0:
-            raise RuntimeError("Failed to fetch model. \
-The path or model %s does not exist." % (uri))
+        s3 = boto3.resource('s3', endpoint_url=os.getenv("AWS_ENDPOINT_URL", "http://s3.amazonaws.com"))
+        parsed = urlparse(uri, scheme='s3')
+        bucket_name = parsed.netloc
+        bucket_path = parsed.path.lstrip('/')
+
+        bucket = s3.Bucket(bucket_name)
+        for obj in bucket.objects.filter(Prefix=bucket_path):
+            # Skip where boto3 lists the directory as an object
+            if obj.key.endswith("/"):
+                continue
+            # In the case where bucket_path points to a single object, set the target key to bucket_path
+            # Otherwise, remove the bucket_path prefix, strip any extra slashes, then prepend the target_dir
+            target_key = (
+                obj.key
+                if bucket_path == obj.key
+                else obj.key.replace(bucket_path, "", 1).lstrip("/")
+            )
+            target = f"{temp_dir}/{target_key}"
+            if not os.path.exists(os.path.dirname(target)):
+                os.makedirs(os.path.dirname(target), exist_ok=True)
+            bucket.download_file(obj.key, target)
 
     @staticmethod
     def _download_gcs(uri, temp_dir: str):
@@ -274,14 +278,3 @@
             os.remove(local_path)
 
         return out_dir
-
-    @staticmethod
-    def _create_minio_client():
-        # Adding prefixing "http" in urlparse is necessary for it to be the netloc
-        url = urlparse(os.getenv("AWS_ENDPOINT_URL", "http://s3.amazonaws.com"))
-        use_ssl = url.scheme == 'https' if url.scheme else bool(os.getenv("S3_USE_HTTPS", "true"))
-        return Minio(url.netloc,
-                     access_key=os.getenv("AWS_ACCESS_KEY_ID", ""),
-                     secret_key=os.getenv("AWS_SECRET_ACCESS_KEY", ""),
-                     region=os.getenv("AWS_REGION", ""),
-                     secure=use_ssl)

{"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@@ -13,22 +13,24 @@\n # limitations under the License.\n \n import glob\n+import gzip\n import logging\n-import tempfile\n import mimetypes\n import os\n import re\n import json\n import shutil\n import tarfile\n+import tempfile\n import zipfile\n-import gzip\n from urllib.parse import urlparse\n+\n+import boto3\n import requests\n from azure.storage.blob import BlockBlobService\n from google.auth import exceptions\n from google.cloud import storage\n-from minio import Minio\n+\n from kfserving.kfmodel_repository import MODEL_MOUNT_DIRS\n \n _GCS_PREFIX = \"gs://\"\n@@ -81,25 +83,27 @@\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-        count = 0\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-            # fget_object handles directory creation if does not exist\n-            if not obj.is_dir:\n-                if subdir_object_key == \"\":\n-                    subdir_object_key = obj.object_name\n-                client.fget_object(bucket_name, obj.object_name,\n-                                   os.path.join(temp_dir, subdir_object_key))\n-            count = count + 1\n-        if count == 0:\n-            raise RuntimeError(\"Failed to fetch model. \\\n-The path or model %s does not exist.\" % (uri))\n+        s3 = boto3.resource('s3', endpoint_url=os.getenv(\"AWS_ENDPOINT_URL\", \"http://s3.amazonaws.com\"))\n+        parsed = urlparse(uri, scheme='s3')\n+        bucket_name = parsed.netloc\n+        bucket_path = parsed.path.lstrip('/')\n+\n+        bucket = s3.Bucket(bucket_name)\n+        for obj in bucket.objects.filter(Prefix=bucket_path):\n+            # Skip where boto3 lists the directory as an object\n+            if obj.key.endswith(\"/\"):\n+                continue\n+            # In the case where bucket_path points to a single object, set the target key to bucket_path\n+            # Otherwise, remove the bucket_path prefix, strip any extra slashes, then prepend the target_dir\n+            target_key = (\n+                obj.key\n+                if bucket_path == obj.key\n+                else obj.key.replace(bucket_path, \"\", 1).lstrip(\"/\")\n+            )\n+            target = f\"{temp_dir}/{target_key}\"\n+            if not os.path.exists(os.path.dirname(target)):\n+                os.makedirs(os.path.dirname(target), exist_ok=True)\n+            bucket.download_file(obj.key, target)\n \n     @staticmethod\n     def _download_gcs(uri, temp_dir: str):\n@@ -274,14 +278,3 @@\n             os.remove(local_path)\n \n         return out_dir\n-\n-    @staticmethod\n-    def _create_minio_client():\n-        # Adding prefixing \"http\" in urlparse is necessary for it to be the netloc\n-        url = urlparse(os.getenv(\"AWS_ENDPOINT_URL\", \"http://s3.amazonaws.com\"))\n-        use_ssl = url.scheme == 'https' if url.scheme else bool(os.getenv(\"S3_USE_HTTPS\", \"true\"))\n-        return Minio(url.netloc,\n-                     access_key=os.getenv(\"AWS_ACCESS_KEY_ID\", \"\"),\n-                     secret_key=os.getenv(\"AWS_SECRET_ACCESS_KEY\", \"\"),\n-                     region=os.getenv(\"AWS_REGION\", \"\"),\n-                     secure=use_ssl)\n", "issue": "[AWS] Support IAM Role for Service Account in KFServing\n/kind feature\r\n\r\n**Describe the solution you'd like**\r\n[A clear and concise description of what you want to happen.]\r\nCurrently, it needs a S3 credential to download model. We need more fine grain control ways. [IRSA](https://docs.aws.amazon.com/eks/latest/userguide/iam-roles-for-service-accounts.html) is a better option. This helps us deploy model without credentials\r\n \r\n**Anything else you would like to add:**\r\n[Miscellaneous information that will assist in solving the issue.]\n", "before_files": [{"content": "# Copyright 2020 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 mimetypes\nimport os\nimport re\nimport json\nimport shutil\nimport tarfile\nimport zipfile\nimport gzip\nfrom urllib.parse import urlparse\nimport requests\nfrom azure.storage.blob import BlockBlobService\nfrom google.auth import exceptions\nfrom google.cloud import storage\nfrom minio import Minio\nfrom kfserving.kfmodel_repository import MODEL_MOUNT_DIRS\n\n_GCS_PREFIX = \"gs://\"\n_S3_PREFIX = \"s3://\"\n_BLOB_RE = \"https://(.+?).blob.core.windows.net/(.+)\"\n_LOCAL_PREFIX = \"file://\"\n_URI_RE = \"https?://(.+)/(.+)\"\n_HTTP_PREFIX = \"http(s)://\"\n_HEADERS_SUFFIX = \"-headers\"\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        elif not os.path.exists(out_dir):\n            os.mkdir(out_dir)\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        elif re.search(_URI_RE, uri):\n            return Storage._download_from_uri(uri, out_dir)\n        elif uri.startswith(MODEL_MOUNT_DIRS):\n            # Don't need to download models if this InferenceService is running in the multi-model\n            # serving mode. The model agent will download models.\n            return out_dir\n        else:\n            raise Exception(\"Cannot recognize storage type for \" + uri +\n                            \"\\n'%s', '%s', '%s', and '%s' are the current available storage type.\" %\n                            (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX, _HTTP_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        count = 0\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            # fget_object handles directory creation if does not exist\n            if not obj.is_dir:\n                if subdir_object_key == \"\":\n                    subdir_object_key = obj.object_name\n                client.fget_object(bucket_name, obj.object_name,\n                                   os.path.join(temp_dir, subdir_object_key))\n            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % (uri))\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        count = 0\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            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % uri)\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 Exception:  # pylint: disable=broad-except\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        count = 0\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            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % (uri))\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 RuntimeError(\"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 _download_from_uri(uri, out_dir=None):\n        url = urlparse(uri)\n        filename = os.path.basename(url.path)\n        mimetype, encoding = mimetypes.guess_type(url.path)\n        local_path = os.path.join(out_dir, filename)\n\n        if filename == '':\n            raise ValueError('No filename contained in URI: %s' % (uri))\n\n        # Get header information from host url\n        headers = {}\n        host_uri = url.hostname\n\n        headers_json = os.getenv(host_uri + _HEADERS_SUFFIX, \"{}\")\n        headers = json.loads(headers_json)\n\n        with requests.get(uri, stream=True, headers=headers) as response:\n            if response.status_code != 200:\n                raise RuntimeError(\"URI: %s returned a %s response code.\" % (uri, response.status_code))\n            if mimetype == 'application/zip' and not response.headers.get('Content-Type', '')\\\n                    .startswith('application/zip'):\n                raise RuntimeError(\"URI: %s did not respond with \\'Content-Type\\': \\'application/zip\\'\" % uri)\n            if mimetype == 'application/x-tar' and not response.headers.get('Content-Type', '')\\\n                    .startswith('application/x-tar'):\n                raise RuntimeError(\"URI: %s did not respond with \\'Content-Type\\': \\'application/x-tar\\'\" % uri)\n            if (mimetype != 'application/zip' and mimetype != 'application/x-tar') and \\\n                    not response.headers.get('Content-Type', '').startswith('application/octet-stream'):\n                raise RuntimeError(\"URI: %s did not respond with \\'Content-Type\\': \\'application/octet-stream\\'\"\n                                   % uri)\n\n            if encoding == 'gzip':\n                stream = gzip.GzipFile(fileobj=response.raw)\n                local_path = os.path.join(out_dir, f'{filename}.tar')\n            else:\n                stream = response.raw\n            with open(local_path, 'wb') as out:\n                shutil.copyfileobj(stream, out)\n\n        if mimetype in [\"application/x-tar\", \"application/zip\"]:\n            if mimetype == \"application/x-tar\":\n                archive = tarfile.open(local_path, 'r', encoding='utf-8')\n            else:\n                archive = zipfile.ZipFile(local_path, 'r')\n            archive.extractall(out_dir)\n            archive.close()\n            os.remove(local_path)\n\n        return out_dir\n\n    @staticmethod\n    def _create_minio_client():\n        # Adding prefixing \"http\" in urlparse is necessary for it to be the netloc\n        url = urlparse(os.getenv(\"AWS_ENDPOINT_URL\", \"http://s3.amazonaws.com\"))\n        use_ssl = url.scheme == 'https' if url.scheme else bool(os.getenv(\"S3_USE_HTTPS\", \"true\"))\n        return Minio(url.netloc,\n                     access_key=os.getenv(\"AWS_ACCESS_KEY_ID\", \"\"),\n                     secret_key=os.getenv(\"AWS_SECRET_ACCESS_KEY\", \"\"),\n                     region=os.getenv(\"AWS_REGION\", \"\"),\n                     secure=use_ssl)\n", "path": "python/kfserving/kfserving/storage.py"}], "after_files": [{"content": "# Copyright 2020 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 gzip\nimport logging\nimport mimetypes\nimport os\nimport re\nimport json\nimport shutil\nimport tarfile\nimport tempfile\nimport zipfile\nfrom urllib.parse import urlparse\n\nimport boto3\nimport requests\nfrom azure.storage.blob import BlockBlobService\nfrom google.auth import exceptions\nfrom google.cloud import storage\n\nfrom kfserving.kfmodel_repository import MODEL_MOUNT_DIRS\n\n_GCS_PREFIX = \"gs://\"\n_S3_PREFIX = \"s3://\"\n_BLOB_RE = \"https://(.+?).blob.core.windows.net/(.+)\"\n_LOCAL_PREFIX = \"file://\"\n_URI_RE = \"https?://(.+)/(.+)\"\n_HTTP_PREFIX = \"http(s)://\"\n_HEADERS_SUFFIX = \"-headers\"\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        elif not os.path.exists(out_dir):\n            os.mkdir(out_dir)\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        elif re.search(_URI_RE, uri):\n            return Storage._download_from_uri(uri, out_dir)\n        elif uri.startswith(MODEL_MOUNT_DIRS):\n            # Don't need to download models if this InferenceService is running in the multi-model\n            # serving mode. The model agent will download models.\n            return out_dir\n        else:\n            raise Exception(\"Cannot recognize storage type for \" + uri +\n                            \"\\n'%s', '%s', '%s', and '%s' are the current available storage type.\" %\n                            (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX, _HTTP_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        s3 = boto3.resource('s3', endpoint_url=os.getenv(\"AWS_ENDPOINT_URL\", \"http://s3.amazonaws.com\"))\n        parsed = urlparse(uri, scheme='s3')\n        bucket_name = parsed.netloc\n        bucket_path = parsed.path.lstrip('/')\n\n        bucket = s3.Bucket(bucket_name)\n        for obj in bucket.objects.filter(Prefix=bucket_path):\n            # Skip where boto3 lists the directory as an object\n            if obj.key.endswith(\"/\"):\n                continue\n            # In the case where bucket_path points to a single object, set the target key to bucket_path\n            # Otherwise, remove the bucket_path prefix, strip any extra slashes, then prepend the target_dir\n            target_key = (\n                obj.key\n                if bucket_path == obj.key\n                else obj.key.replace(bucket_path, \"\", 1).lstrip(\"/\")\n            )\n            target = f\"{temp_dir}/{target_key}\"\n            if not os.path.exists(os.path.dirname(target)):\n                os.makedirs(os.path.dirname(target), exist_ok=True)\n            bucket.download_file(obj.key, target)\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        count = 0\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            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % uri)\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 Exception:  # pylint: disable=broad-except\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        count = 0\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            count = count + 1\n        if count == 0:\n            raise RuntimeError(\"Failed to fetch model. \\\nThe path or model %s does not exist.\" % (uri))\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 RuntimeError(\"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 _download_from_uri(uri, out_dir=None):\n        url = urlparse(uri)\n        filename = os.path.basename(url.path)\n        mimetype, encoding = mimetypes.guess_type(url.path)\n        local_path = os.path.join(out_dir, filename)\n\n        if filename == '':\n            raise ValueError('No filename contained in URI: %s' % (uri))\n\n        # Get header information from host url\n        headers = {}\n        host_uri = url.hostname\n\n        headers_json = os.getenv(host_uri + _HEADERS_SUFFIX, \"{}\")\n        headers = json.loads(headers_json)\n\n        with requests.get(uri, stream=True, headers=headers) as response:\n            if response.status_code != 200:\n                raise RuntimeError(\"URI: %s returned a %s response code.\" % (uri, response.status_code))\n            if mimetype == 'application/zip' and not response.headers.get('Content-Type', '')\\\n                    .startswith('application/zip'):\n                raise RuntimeError(\"URI: %s did not respond with \\'Content-Type\\': \\'application/zip\\'\" % uri)\n            if mimetype == 'application/x-tar' and not response.headers.get('Content-Type', '')\\\n                    .startswith('application/x-tar'):\n                raise RuntimeError(\"URI: %s did not respond with \\'Content-Type\\': \\'application/x-tar\\'\" % uri)\n            if (mimetype != 'application/zip' and mimetype != 'application/x-tar') and \\\n                    not response.headers.get('Content-Type', '').startswith('application/octet-stream'):\n                raise RuntimeError(\"URI: %s did not respond with \\'Content-Type\\': \\'application/octet-stream\\'\"\n                                   % uri)\n\n            if encoding == 'gzip':\n                stream = gzip.GzipFile(fileobj=response.raw)\n                local_path = os.path.join(out_dir, f'{filename}.tar')\n            else:\n                stream = response.raw\n            with open(local_path, 'wb') as out:\n                shutil.copyfileobj(stream, out)\n\n        if mimetype in [\"application/x-tar\", \"application/zip\"]:\n            if mimetype == \"application/x-tar\":\n                archive = tarfile.open(local_path, 'r', encoding='utf-8')\n            else:\n                archive = zipfile.ZipFile(local_path, 'r')\n            archive.extractall(out_dir)\n            archive.close()\n            os.remove(local_path)\n\n        return out_dir\n", "path": "python/kfserving/kfserving/storage.py"}]}