Add AssumeRoleWithCertificate credential provider. (#1182)

Signed-off-by: Bala.FA <[email protected]>

examples/minio_with_certificate_identity_provider.py

@@ -0,0 +1,38 @@
+# -*- coding: utf-8 -*-
+# MinIO Python Library for Amazon S3 Compatible Cloud Storage,
+# (C) 2022 MinIO, Inc.
+#
+# Licensed under the Apache License, Version 2.0 (the "License");
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an "AS IS" BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+#
+
+from minio import Minio
+from minio.credentials import CertificateIdentityProvider
+
+# STS endpoint usually point to MinIO server.
+sts_endpoint = "https://STS-HOST:STS-PORT/"
+
+# client certificate file
+cert_file = "/path/to/client.pem"
+
+# client private key
+key_file = "/path/to/client.key"
+
+provider = CertificateIdentityProvider(
+ sts_endpoint, cert_file=cert_file, key_file=key_file,
+)
+
+client = Minio("MINIO-HOST:MINIO-PORT", credentials=provider)
+
+# Get information of an object.
+stat = client.stat_object("my-bucket", "my-object")
+print(stat)

minio/credentials/providers.py

@@ -29,6 +29,7 @@
 from urllib.parse import urlencode, urlsplit
 from xml.etree import ElementTree
 
+import certifi
 import urllib3
 
 from minio.helpers import sha256_hash
@@ -593,3 +594,66 @@ class WebIdentityProvider(WebIdentityClientGrantsProvider):
 
  def _is_web_identity(self):
  return True
+
+
+class CertificateIdentityProvider(Provider):
+ """Credential provider using AssumeRoleWithCertificate API."""
+
+ def __init__(
+ self, sts_endpoint, cert_file=None, key_file=None,
+ key_password=None, ca_certs=None, duration_seconds=0,
+ http_client=None,
+ ):
+ if urlsplit(sts_endpoint).scheme != "https":
+ raise ValueError("STS endpoint scheme must be HTTPS")
+
+ if bool(http_client) != (cert_file and key_file):
+ pass
+ else:
+ raise ValueError(
+ "either cert/key file or custom http_client must be provided",
+ )
+
+ self._sts_endpoint = sts_endpoint + "?" + urlencode(
+ {
+ "Action": "AssumeRoleWithCertificate",
+ "Version": "2011-06-15",
+ "DurationSeconds": str(
+ duration_seconds
+ if duration_seconds > _DEFAULT_DURATION_SECONDS
+ else _DEFAULT_DURATION_SECONDS
+ ),
+ },
+ )
+ self._http_client = http_client or urllib3.PoolManager(
+ maxsize=10,
+ cert_file=cert_file,
+ cert_reqs='CERT_REQUIRED',
+ key_file=key_file,
+ key_password=key_password,
+ ca_certs=ca_certs or certifi.where(),
+ retries=urllib3.Retry(
+ total=5,
+ backoff_factor=0.2,
+ status_forcelist=[500, 502, 503, 504],
+ ),
+ )
+ self._credentials = None
+
+ def retrieve(self):
+ """Retrieve credentials."""
+
+ if self._credentials and not self._credentials.is_expired():
+ return self._credentials
+
+ res = _urlopen(
+ self._http_client,
+ "POST",
+ self._sts_endpoint,
+ )
+
+ self._credentials = _parse_credentials(
+ res.data.decode(), "AssumeRoleWithCertificateResult",
+ )
+
+ return self._credentials