We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Support insecure minio connections
/kind feature
For local deployments (e.g. with Kubeflow), TLS is not configured for minio. When trying to deploy a model stored there, kfserving issues SSLError messages.

```Caused by SSLError(SSLError(1, '[SSL: WRONG_VERSION_NUMBER] wrong version number (_ssl.c:1056)')```

This is due to a hard-coded setting in [storage.py](https://github.com/kubeflow/kfserving/blob/master/python/kfserving/kfserving/storage.py#L194), with no way to override it.

**Describe the solution you'd like**
It should be possible to allow insecure connections to minio, having secure connection as default. The user should be able to do this either by specifying an S3_ENDPOINT with http scheme or by using a flag explicitly.


--- 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.common import AzureMissingResourceHttpError
21 from azure.storage.blob import BlockBlobService
22 from google.auth import exceptions
23 from google.cloud import storage
24 from minio import Minio
25 
26 _GCS_PREFIX = "gs://"
27 _S3_PREFIX = "s3://"
28 _BLOB_RE = "https://(.+?).blob.core.windows.net/(.+)"
29 _LOCAL_PREFIX = "file://"
30 
31 
32 class Storage(object): # pylint: disable=too-few-public-methods
33     @staticmethod
34     def download(uri: str, out_dir: str = None) -> str:
35         logging.info("Copying contents of %s to local", uri)
36 
37         is_local = False
38         if uri.startswith(_LOCAL_PREFIX) or os.path.exists(uri):
39             is_local = True
40 
41         if out_dir is None:
42             if is_local:
43                 # noop if out_dir is not set and the path is local
44                 return Storage._download_local(uri)
45             out_dir = tempfile.mkdtemp()
46 
47         if uri.startswith(_GCS_PREFIX):
48             Storage._download_gcs(uri, out_dir)
49         elif uri.startswith(_S3_PREFIX):
50             Storage._download_s3(uri, out_dir)
51         elif re.search(_BLOB_RE, uri):
52             Storage._download_blob(uri, out_dir)
53         elif is_local:
54             return Storage._download_local(uri, out_dir)
55         else:
56             raise Exception("Cannot recognize storage type for " + uri +
57                             "\n'%s', '%s', and '%s' are the current available storage type." %
58                             (_GCS_PREFIX, _S3_PREFIX, _LOCAL_PREFIX))
59 
60         logging.info("Successfully copied %s to %s", uri, out_dir)
61         return out_dir
62 
63     @staticmethod
64     def _download_s3(uri, temp_dir: str):
65         client = Storage._create_minio_client()
66         bucket_args = uri.replace(_S3_PREFIX, "", 1).split("/", 1)
67         bucket_name = bucket_args[0]
68         bucket_path = bucket_args[1] if len(bucket_args) > 1 else ""
69         objects = client.list_objects(bucket_name, prefix=bucket_path, recursive=True)
70         for obj in objects:
71             # Replace any prefix from the object key with temp_dir
72             subdir_object_key = obj.object_name.replace(bucket_path, "", 1).strip("/")
73             if subdir_object_key == "":
74                 subdir_object_key = obj.object_name.split("/")[-1]
75             client.fget_object(bucket_name, obj.object_name,
76                                os.path.join(temp_dir, subdir_object_key))
77 
78     @staticmethod
79     def _download_gcs(uri, temp_dir: str):
80         try:
81             storage_client = storage.Client()
82         except exceptions.DefaultCredentialsError:
83             storage_client = storage.Client.create_anonymous_client()
84         bucket_args = uri.replace(_GCS_PREFIX, "", 1).split("/", 1)
85         bucket_name = bucket_args[0]
86         bucket_path = bucket_args[1] if len(bucket_args) > 1 else ""
87         bucket = storage_client.bucket(bucket_name)
88         prefix = bucket_path
89         if not prefix.endswith("/"):
90             prefix = prefix + "/"
91         blobs = bucket.list_blobs(prefix=prefix)
92         for blob in blobs:
93             # Replace any prefix from the object key with temp_dir
94             subdir_object_key = blob.name.replace(bucket_path, "", 1).strip("/")
95 
96             # Create necessary subdirectory to store the object locally
97             if "/" in subdir_object_key:
98                 local_object_dir = os.path.join(temp_dir, subdir_object_key.rsplit("/", 1)[0])
99                 if not os.path.isdir(local_object_dir):
100                     os.makedirs(local_object_dir, exist_ok=True)
101             if subdir_object_key.strip() != "":
102                 dest_path = os.path.join(temp_dir, subdir_object_key)
103                 logging.info("Downloading: %s", dest_path)
104                 blob.download_to_filename(dest_path)
105 
106     @staticmethod
107     def _download_blob(uri, out_dir: str): # pylint: disable=too-many-locals
108         match = re.search(_BLOB_RE, uri)
109         account_name = match.group(1)
110         storage_url = match.group(2)
111         container_name, prefix = storage_url.split("/", 1)
112 
113         logging.info("Connecting to BLOB account: [%s], container: [%s], prefix: [%s]",
114                      account_name,
115                      container_name,
116                      prefix)
117         try:
118             block_blob_service = BlockBlobService(account_name=account_name)
119             blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
120         except AzureMissingResourceHttpError:
121             token = Storage._get_azure_storage_token()
122             if token is None:
123                 logging.warning("Azure credentials not found, retrying anonymous access")
124             block_blob_service = BlockBlobService(account_name=account_name, token_credential=token)
125             blobs = block_blob_service.list_blobs(container_name, prefix=prefix)
126 
127         for blob in blobs:
128             dest_path = os.path.join(out_dir, blob.name)
129             if "/" in blob.name:
130                 head, tail = os.path.split(blob.name)
131                 if prefix is not None:
132                     head = head[len(prefix):]
133                 if head.startswith('/'):
134                     head = head[1:]
135                 dir_path = os.path.join(out_dir, head)
136                 dest_path = os.path.join(dir_path, tail)
137                 if not os.path.isdir(dir_path):
138                     os.makedirs(dir_path)
139 
140             logging.info("Downloading: %s to %s", blob.name, dest_path)
141             block_blob_service.get_blob_to_path(container_name, blob.name, dest_path)
142 
143     @staticmethod
144     def _get_azure_storage_token():
145         tenant_id = os.getenv("AZ_TENANT_ID", "")
146         client_id = os.getenv("AZ_CLIENT_ID", "")
147         client_secret = os.getenv("AZ_CLIENT_SECRET", "")
148         subscription_id = os.getenv("AZ_SUBSCRIPTION_ID", "")
149 
150         if tenant_id == "" or client_id == "" or client_secret == "" or subscription_id == "":
151             return None
152 
153         # note the SP must have "Storage Blob Data Owner" perms for this to work
154         import adal
155         from azure.storage.common import TokenCredential
156 
157         authority_url = "https://login.microsoftonline.com/" + tenant_id
158 
159         context = adal.AuthenticationContext(authority_url)
160 
161         token = context.acquire_token_with_client_credentials(
162             "https://storage.azure.com/",
163             client_id,
164             client_secret)
165 
166         token_credential = TokenCredential(token["accessToken"])
167 
168         logging.info("Retrieved SP token credential for client_id: %s", client_id)
169 
170         return token_credential
171 
172     @staticmethod
173     def _download_local(uri, out_dir=None):
174         local_path = uri.replace(_LOCAL_PREFIX, "", 1)
175         if not os.path.exists(local_path):
176             raise Exception("Local path %s does not exist." % (uri))
177 
178         if out_dir is None:
179             return local_path
180         elif not os.path.isdir(out_dir):
181             os.makedirs(out_dir)
182 
183         if os.path.isdir(local_path):
184             local_path = os.path.join(local_path, "*")
185 
186         for src in glob.glob(local_path):
187             _, tail = os.path.split(src)
188             dest_path = os.path.join(out_dir, tail)
189             logging.info("Linking: %s to %s", src, dest_path)
190             os.symlink(src, dest_path)
191         return out_dir
192 
193     @staticmethod
194     def _create_minio_client():
195         # Remove possible http scheme for Minio
196         url = re.compile(r"https?://")
197         minioClient = Minio(url.sub("", os.getenv("S3_ENDPOINT", "")),
198                             access_key=os.getenv("AWS_ACCESS_KEY_ID", ""),
199                             secret_key=os.getenv("AWS_SECRET_ACCESS_KEY", ""),
200                             secure=True)
201         return minioClient
202
```
--- 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 urllib.parse import urlparse
 from azure.common import AzureMissingResourceHttpError
 from azure.storage.blob import BlockBlobService
 from google.auth import exceptions
@@ -193,9 +194,9 @@
     @staticmethod
     def _create_minio_client():
         # Remove possible http scheme for Minio
-        url = re.compile(r"https?://")
-        minioClient = Minio(url.sub("", os.getenv("S3_ENDPOINT", "")),
-                            access_key=os.getenv("AWS_ACCESS_KEY_ID", ""),
-                            secret_key=os.getenv("AWS_SECRET_ACCESS_KEY", ""),
-                            secure=True)
-        return minioClient
+        url = urlparse(os.getenv("AWS_ENDPOINT_URL", ""))
+        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", ""),
+                     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@@ -17,6 +17,7 @@\n import tempfile\n import os\n import re\n+from urllib.parse import urlparse\n from azure.common import AzureMissingResourceHttpError\n from azure.storage.blob import BlockBlobService\n from google.auth import exceptions\n@@ -193,9 +194,9 @@\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+        url = urlparse(os.getenv(\"AWS_ENDPOINT_URL\", \"\"))\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+                     secure=use_ssl)\n", "issue": "Support insecure minio connections\n/kind feature\r\nFor local deployments (e.g. with Kubeflow), TLS is not configured for minio. When trying to deploy a model stored there, kfserving issues SSLError messages.\r\n\r\n```Caused by SSLError(SSLError(1, '[SSL: WRONG_VERSION_NUMBER] wrong version number (_ssl.c:1056)')```\r\n\r\nThis is due to a hard-coded setting in [storage.py](https://github.com/kubeflow/kfserving/blob/master/python/kfserving/kfserving/storage.py#L194), with no way to override it.\r\n\r\n**Describe the solution you'd like**\r\nIt should be possible to allow insecure connections to minio, having secure connection as default. The user should be able to do this either by specifying an S3_ENDPOINT with http scheme or by using a flag explicitly.\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.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            if subdir_object_key == \"\":\n                subdir_object_key = obj.object_name.split(\"/\")[-1]\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"}], "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 urllib.parse import urlparse\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            if subdir_object_key == \"\":\n                subdir_object_key = obj.object_name.split(\"/\")[-1]\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 = urlparse(os.getenv(\"AWS_ENDPOINT_URL\", \"\"))\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                     secure=use_ssl)\n", "path": "python/kfserving/kfserving/storage.py"}]}