Refactor MinioAdminClient using HTTP client

.github/workflows/ci.yml

@@ -45,7 +45,7 @@ jobs:
     - name: Install dependencies
       run: |
         python -m pip install --upgrade pip setuptools
-        pip install certifi urllib3 mock pytest
+        pip install certifi urllib3 mock pytest argon2-cffi pycryptodome
     - name: Run check if Ubuntu
       if: matrix.os == 'ubuntu-latest'
       run: |

Makefile

@@ -3,7 +3,7 @@ default: tests
 
 getdeps:
 	@echo "Installing required dependencies"
-	@pip install --user --upgrade autopep8 certifi pytest pylint urllib3
+	@pip install --user --upgrade autopep8 certifi pytest pylint urllib3 argon2-cffi pycryptodome
 
 check: getdeps
 	@echo "Running checks"

minio/api.py

@@ -25,7 +25,6 @@
 
 import itertools
 import os
-import platform
 import tarfile
 from datetime import timedelta
 from io import BytesIO
@@ -47,10 +46,11 @@
                         parse_copy_object, parse_list_objects)
 from .deleteobjects import DeleteError, DeleteRequest, DeleteResult
 from .error import InvalidResponseError, S3Error, ServerError
-from .helpers import (MAX_MULTIPART_COUNT, MAX_MULTIPART_OBJECT_SIZE,
-                      MAX_PART_SIZE, MIN_PART_SIZE, BaseURL, ObjectWriteResult,
-                      ThreadPool, check_bucket_name, check_non_empty_string,
-                      check_sse, check_ssec, genheaders, get_part_info,
+from .helpers import (_DEFAULT_USER_AGENT, MAX_MULTIPART_COUNT,
+                      MAX_MULTIPART_OBJECT_SIZE, MAX_PART_SIZE, MIN_PART_SIZE,
+                      BaseURL, ObjectWriteResult, ThreadPool,
+                      check_bucket_name, check_non_empty_string, check_sse,
+                      check_ssec, genheaders, get_part_info,
                       headers_to_strings, is_valid_policy_type, makedirs,
                       md5sum_hash, read_part_data, sha256_hash)
 from .legalhold import LegalHold
@@ -67,11 +67,6 @@
 from .versioningconfig import VersioningConfig
 from .xml import Element, SubElement, findtext, getbytes, marshal, unmarshal
 
-_DEFAULT_USER_AGENT = (
-    f"MinIO ({platform.system()}; {platform.machine()}) "
-    f"{__title__}/{__version__}"
-)
-
 
 class Minio:  # pylint: disable=too-many-public-methods
     """

minio/crypto.py

@@ -0,0 +1,146 @@
+# -*- coding: utf-8 -*-
+# MinIO Python Library for Amazon S3 Compatible Cloud Storage, (C)
+# 2015, 2016, 2017 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.
+
+# pylint: disable=too-many-lines,disable=too-many-branches,too-many-statements
+# pylint: disable=too-many-arguments
+
+"""Cryptography to read and write encrypted MinIO Admin payload"""
+
+import os
+
+from argon2.low_level import Type, hash_secret_raw
+from Crypto.Cipher import AES, ChaCha20_Poly1305
+
+_NONCE_LEN = 8
+_SALT_LEN = 32
+
+
+class AesGcmCipherProvider:
+    """AES-GCM cipher provider"""
+    @staticmethod
+    def get_cipher(key: bytes, nonce: bytes):
+        """Get cipher"""
+        return AES.new(key, AES.MODE_GCM, nonce)
+
+
+class ChaCha20Poly1305CipherProvider:
+    """ChaCha20Poly1305 cipher provider"""
+    @staticmethod
+    def get_cipher(key: bytes, nonce: bytes):
+        """Get cipher"""
+        return ChaCha20_Poly1305.new(key=key, nonce=nonce)
+
+
+def encrypt(payload: bytes, password: str) -> bytes:
+    """
+    Encrypts data using AES-GCM using a 256-bit Argon2ID key.
+    To see the original implementation in Go, check out the madmin-go library
+    (https://github.com/minio/madmin-go/blob/main/encrypt.go#L38)
+    """
+    cipher_provider = AesGcmCipherProvider()
+    nonce = os.urandom(_NONCE_LEN)
+    salt = os.urandom(_SALT_LEN)
+
+    padded_nonce = [0] * (_NONCE_LEN + 4)
+    padded_nonce[:_NONCE_LEN] = nonce
+
+    key = _generate_key(password.encode(), salt)
+    additional_data = _generate_additional_data(
+        cipher_provider, key, bytes(padded_nonce))
+
+    padded_nonce[8] = 0x01
+    padded_nonce = bytes(padded_nonce)
+
+    cipher = cipher_provider.get_cipher(key, padded_nonce)
+    cipher.update(additional_data)
+    encrypted_data, mac = cipher.encrypt_and_digest(payload)
+
+    payload = salt
+    payload += bytes([0x00])
+    payload += nonce
+    payload += encrypted_data
+    payload += mac
+
+    return bytes(payload)
+
+
+def decrypt(payload: bytes, password: str) -> bytes:
+    """
+    Decrypts data using AES-GCM or ChaCha20Poly1305 using a
+    256-bit Argon2ID key. To see the original implementation in Go,
+    check out the madmin-go library
+    (https://github.com/minio/madmin-go/blob/main/encrypt.go#L38)
+    """
+    pos = 0
+    salt = payload[pos:pos+_SALT_LEN]
+    pos += _SALT_LEN
+
+    cipher_id = payload[pos]
+    if cipher_id == 0:
+        cipher_provider = AesGcmCipherProvider()
+    elif cipher_id == 1:
+        cipher_provider = ChaCha20Poly1305CipherProvider()
+    else:
+        return None
+
+    pos += 1
+
+    nonce = payload[pos:pos+_NONCE_LEN]
+    pos += _NONCE_LEN
+
+    encrypted_data = payload[pos:-16]
+    hmac_tag = payload[-16:]
+
+    key = _generate_key(password.encode(), salt)
+
+    padded_nonce = [0] * 12
+    padded_nonce[:_NONCE_LEN] = nonce
+
+    additional_data = _generate_additional_data(
+        cipher_provider, key, bytes(padded_nonce))
+    padded_nonce[8] = 1
+
+    cipher = cipher_provider.get_cipher(key, bytes(padded_nonce))
+
+    cipher.update(additional_data)
+    decrypted_data = cipher.decrypt_and_verify(encrypted_data, hmac_tag)
+
+    return decrypted_data
+
+
+def _generate_additional_data(cipher_provider, key: bytes,
+                              padded_nonce: bytes) -> bytes:
+    """Generate additional data"""
+    cipher = cipher_provider.get_cipher(key, padded_nonce)
+    tag = cipher.digest()
+    new_tag = [0] * 17
+    new_tag[1:] = tag
+    new_tag[0] = 0x80
+    return bytes(new_tag)
+
+
+def _generate_key(password: bytes, salt: bytes) -> bytes:
+    """Generate 256-bit Argon2ID key"""
+    return hash_secret_raw(
+        secret=password,
+        salt=salt,
+        time_cost=1,
+        memory_cost=65536,
+        parallelism=4,
+        hash_len=32,
+        type=Type.ID,
+        version=19
+    )

minio/error.py

@@ -144,3 +144,17 @@ def copy(self, code, message):
             self._bucket_name,
             self._object_name,
         )
+
+
+class MinioAdminException(Exception):
+    """Raised to indicate admin API execution error."""
+
+    def __init__(self, code, body):
+        self._code = code
+        self._body = body
+        super().__init__(
+            f"admin request failed; Status: {code}, Body: {body}",
+        )
+
+    def __reduce__(self):
+        return type(self), (self._code, self._body)

minio/helpers.py

@@ -23,15 +23,22 @@
 import hashlib
 import math
 import os
+import platform
 import re
 import urllib.parse
 from queue import Queue
 from threading import BoundedSemaphore, Thread
 
+from . import __title__, __version__
 from .sse import Sse, SseCustomerKey
 from .time import to_iso8601utc
 
 # Constants
+_DEFAULT_USER_AGENT = (
+    f"MinIO ({platform.system()}; {platform.machine()}) "
+    f"{__title__}/{__version__}"
+)
+
 MAX_MULTIPART_COUNT = 10000  # 10000 parts
 MAX_MULTIPART_OBJECT_SIZE = 5 * 1024 * 1024 * 1024 * 1024  # 5TiB
 MAX_PART_SIZE = 5 * 1024 * 1024 * 1024  # 5GiB
@@ -427,53 +434,61 @@ def _get_aws_info(host, https, region):
              "dualstack": dualstack}, None)
 
 
-class BaseURL:
-    """Base URL of S3 endpoint."""
+def _parse_url(endpoint):
+    """Parse url string."""
 
-    def __init__(self, endpoint, region):
-        url = urllib.parse.urlsplit(endpoint)
-        host = url.hostname
+    url = urllib.parse.urlsplit(endpoint)
+    host = url.hostname
 
-        if url.scheme.lower() not in ["http", "https"]:
-            raise ValueError("scheme in endpoint must be http or https")
+    if url.scheme.lower() not in ["http", "https"]:
+        raise ValueError("scheme in endpoint must be http or https")
 
-        url = url_replace(url, scheme=url.scheme.lower())
+    url = url_replace(url, scheme=url.scheme.lower())
 
-        if url.path and url.path != "/":
-            raise ValueError("path in endpoint is not allowed")
+    if url.path and url.path != "/":
+        raise ValueError("path in endpoint is not allowed")
 
-        url = url_replace(url, path="")
+    url = url_replace(url, path="")
 
-        if url.query:
-            raise ValueError("query in endpoint is not allowed")
+    if url.query:
+        raise ValueError("query in endpoint is not allowed")
 
-        if url.fragment:
-            raise ValueError("fragment in endpoint is not allowed")
+    if url.fragment:
+        raise ValueError("fragment in endpoint is not allowed")
 
-        try:
-            url.port
-        except ValueError as exc:
-            raise ValueError("invalid port") from exc
+    try:
+        url.port
+    except ValueError as exc:
+        raise ValueError("invalid port") from exc
 
-        if url.username:
-            raise ValueError("username in endpoint is not allowed")
+    if url.username:
+        raise ValueError("username in endpoint is not allowed")
 
-        if url.password:
-            raise ValueError("password in endpoint is not allowed")
+    if url.password:
+        raise ValueError("password in endpoint is not allowed")
 
-        if (
-                (url.scheme == "http" and url.port == 80) or
-                (url.scheme == "https" and url.port == 443)
-        ):
-            url = url_replace(url, netloc=host)
+    if (
+            (url.scheme == "http" and url.port == 80) or
+            (url.scheme == "https" and url.port == 443)
+    ):
+        url = url_replace(url, netloc=host)
+
+    return url
+
+
+class BaseURL:
+    """Base URL of S3 endpoint."""
+
+    def __init__(self, endpoint, region):
+        url = _parse_url(endpoint)
 
         if region and not _REGION_REGEX.match(region):
             raise ValueError(f"invalid region {region}")
 
         self._aws_info, region_in_host = _get_aws_info(
-            host, url.scheme == "https", region)
+            url.hostname, url.scheme == "https", region)
         self._virtual_style_flag = (
-            self._aws_info or host.endswith("aliyuncs.com")
+            self._aws_info or url.hostname.endswith("aliyuncs.com")
         )
         self._url = url
         self._region = region or region_in_host