Basic MinIO user management API

minio/__init__.py

@@ -41,3 +41,4 @@
 from .api import Minio
 from .error import InvalidResponseError, S3Error, ServerError
 from .minioadmin import MinioAdmin
+from .minioadminhttp import MinioAdminHttp

minio/datatypes.py

@@ -24,6 +24,7 @@
 import datetime
 import json
 from collections import OrderedDict
+from typing import List
 from urllib.parse import unquote_plus
 from xml.etree import ElementTree as ET
 
@@ -850,3 +851,60 @@ def __enter__(self):
 
     def __exit__(self, exc_type, value, traceback):
         self._close_response()
+
+
+class UserInfo():
+    """MinIO user information."""
+
+    def __init__(self, status: str, member_of: List[str], policies: List[str]):
+        self._status = status
+        self._member_of = member_of
+        self._policies = policies
+
+    @property
+    def status(self):
+        """Get status"""
+        return self._status
+
+    @property
+    def member_of(self):
+        """Get list of groups user is member of"""
+        return self._member_of
+
+    @property
+    def policies(self):
+        """Get list of policies attached to user"""
+        return self._policies
+
+    def __repr__(self):
+        return f"{type(self).__name__}()"
+
+    @classmethod
+    def fromjson(cls, data: str):
+        """Create new object with values from JSON string"""
+        parsed_data = json.loads(data)
+        return cls.fromdict(parsed_data)
+
+    @classmethod
+    def fromdict(cls, data: dict):
+        """Create new object with values from python dictionary"""
+        return cls(
+            data['status'],
+            data.get('memberOf', []),
+            _parse_policies(data.get('policyName', ''))
+        )
+
+
+def _parse_policies(policy_string: str) -> List[str]:
+    """Parse string of policies to list"""
+    policies = policy_string.split(',')
+    return [p for p in policies if p]
+
+
+def parse_list_users(data: str) -> dict:
+    """Parsa data returned from list-users"""
+    json_data = json.loads(data)
+    result = {}
+    for access_key, user_data in json_data.items():
+        result[access_key] = UserInfo.fromdict(user_data)
+    return result

minio/error.py

@@ -53,6 +53,22 @@ def __reduce__(self):
         return type(self), (self._code, self._content_type, self._body)
 
 
+class AdminResponseError(MinioException):
+    """Raised to indicate that non-OK response from server."""
+
+    def __init__(self, code, content_type, body):
+        self._code = code
+        self._content_type = content_type
+        self._body = body
+        super().__init__(
+            f"non-OK admin response from server; "
+            f"Response code: {code}, Body: {body}"
+        )
+
+    def __reduce__(self):
+        return type(self), (self._code, self._content_type, self._body)
+
+
 class ServerError(MinioException):
     """Raised to indicate that S3 service returning HTTP server error."""
 

minio/minioadminhttp.py

@@ -0,0 +1,108 @@
+# -*- coding: utf-8 -*-
+# MinIO Python Library for Amazon S3 Compatible Cloud Storage,
+# (C) 2021 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-public-methods
+
+"""MinIO Admin wrapper using HTTP API."""
+
+from __future__ import absolute_import
+import json
+from typing import Dict
+from minio.api import HttpClient
+from minio.crypto import decrypt, encrypt
+from minio.datatypes import UserInfo, parse_list_users
+from minio.error import AdminResponseError
+
+
+_ADMIN_PATH_PREFIX = "/minio/admin/v3"
+
+
+class MinioAdminHttp(HttpClient):
+    """MinIO Admin wrapper using HTTP API"""
+
+    def __init__(self, endpoint, access_key,
+                 secret_key,
+                 session_token=None,
+                 secure=True,
+                 region="us-east-1",
+                 http_client=None,
+                 credentials=None,
+                 cert_check=True):
+        super().__init__(endpoint, access_key, secret_key, session_token,
+                         secure, region, http_client, credentials, cert_check)
+
+    def _url_open(  # pylint: disable=too-many-branches
+            self,
+            method,
+            path,
+            body=None,
+            headers=None,
+            query_params=None,
+    ):
+        """Execute HTTP request."""
+        url = self._base_url.build(
+            method,
+            region=self._base_url.region,
+            bucket_name=None,
+            object_name=None,
+            query_params=query_params,
+            path=_ADMIN_PATH_PREFIX + path
+        )
+
+        response = self._send_request(
+            method,
+            url,
+            headers,
+            body,
+            region=self._base_url.region
+        )
+
+        if response.status in [200, 204, 206]:
+            return response
+
+        raise AdminResponseError(
+            response.status,
+            response.headers.get("content-type"),
+            response.data.decode() if response.data else None
+        )
+
+    def user_add(self, access_key: str, secret_key: str):
+        """Create user with access and secret keys"""
+        params = {"accessKey": access_key}
+        data = {"secretKey": secret_key}
+        data = json.dumps(data).encode('utf-8')
+        creds = self._provider.retrieve()
+        data = encrypt(data, creds.secret_key)
+        self._url_open("PUT", "/add-user", query_params=params, body=data)
+
+    def user_info(self, access_key: str) -> UserInfo:
+        """Get information about user"""
+        params = {"accessKey": access_key}
+        response = self._url_open("GET", "/user-info", query_params=params)
+        data = response.data.decode()
+        return UserInfo.fromjson(data)
+
+    def list_users(self) -> Dict[str, UserInfo]:
+        """List all users"""
+        response = self._url_open("GET", "/list-users")
+        creds = self._provider.retrieve()
+        data = decrypt(response.data, creds.secret_key).decode()
+        return parse_list_users(data)
+
+    def user_remove(self, access_key: str):
+        """Delete user"""
+        params = {"accessKey": access_key}
+        self._url_open("DELETE", "/remove-user", query_params=params)

tests/unit/list_users_test.py

@@ -0,0 +1,78 @@
+# -*- coding: utf-8 -*-
+# MinIO Python Library for Amazon S3 Compatible Cloud Storage,
+# (C) 2015 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.
+
+import json
+import unittest.mock as mock
+from unittest import TestCase
+
+from minio.api import _DEFAULT_USER_AGENT
+from minio.crypto import encrypt
+from minio.minioadminhttp import MinioAdminHttp
+
+from .minio_mocks import MockConnection, MockResponse
+
+
+class ListUsersTest(TestCase):
+    @mock.patch('urllib3.PoolManager')
+    def test_empty_list_users_works(self, mock_connection):
+        access_key = "minioadmin"
+        secret_key = "minioadmin"
+        mock_data = encrypt(json.dumps({}).encode(), secret_key)
+        mock_server = MockConnection()
+        mock_connection.return_value = mock_server
+        mock_server.mock_add_request(
+            MockResponse('GET', 'https://localhost:9000/minio/admin/v3/list-users',
+                         {'User-Agent': _DEFAULT_USER_AGENT},
+                         200, content=mock_data)
+        )
+        client = MinioAdminHttp('localhost:9000', access_key, secret_key)
+        users = client.list_users()
+        self.assertEqual(0, len(users))
+
+    @mock.patch('urllib3.PoolManager')
+    def test_list_users_works(self, mock_connection):
+        access_key = "minioadmin"
+        secret_key = "minioadmin"
+        users = {
+            'john': {
+                'status': 'enabled',
+                'memberOf': ['group', 'group2'],
+                'policyName': 'policyA,policyB'
+            },
+            'matt': {
+                'status': 'disabled',
+                'memberOf': ['group2', 'group1', 'group3'],
+                'policyName': ''
+            }
+        }
+        mock_data = encrypt(json.dumps(users).encode(), secret_key)
+        mock_server = MockConnection()
+        mock_connection.return_value = mock_server
+        mock_server.mock_add_request(
+            MockResponse('GET', 'https://localhost:9000/minio/admin/v3/list-users',
+                         {'User-Agent': _DEFAULT_USER_AGENT},
+                         200, content=mock_data)
+        )
+        client = MinioAdminHttp('localhost:9000', access_key, secret_key)
+        users = client.list_users()
+        self.assertEqual(2, len(users))
+        self.assertEqual(2, len(users['john'].policies))
+        self.assertEqual(0, len(users['matt'].policies))
+        self.assertEqual(3, len(users['matt'].member_of))
+
+        self.assertEqual('disabled', users['matt'].status)
+        self.assertEqual('policyB', users['john'].policies[1])
+        self.assertEqual('group1', users['matt'].member_of[1])

tests/unit/user_info_test.py

@@ -0,0 +1,49 @@
+# -*- coding: utf-8 -*-
+# MinIO Python Library for Amazon S3 Compatible Cloud Storage,
+# (C) 2015 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.
+
+import json
+import unittest.mock as mock
+from unittest import TestCase
+
+from minio.api import _DEFAULT_USER_AGENT
+from minio.minioadminhttp import MinioAdminHttp
+
+from .minio_mocks import MockConnection, MockResponse
+
+
+class UserInfoTest(TestCase):
+    @mock.patch('urllib3.PoolManager')
+    def test_user_info_works(self, mock_connection):
+        access_key = "minioadmin"
+        secret_key = "minioadmin"
+        user = {
+            'status': 'enabled',
+            'memberOf': ['group', 'group2'],
+            'policyName': 'policyA,policyB'
+        }
+        mock_data = json.dumps(user).encode()
+        mock_server = MockConnection()
+        mock_connection.return_value = mock_server
+        mock_server.mock_add_request(
+            MockResponse('GET', 'https://localhost:9000/minio/admin/v3/user-info?accessKey=user',
+                         {'User-Agent': _DEFAULT_USER_AGENT},
+                         200, content=mock_data)
+        )
+        client = MinioAdminHttp('localhost:9000', access_key, secret_key)
+        user = client.user_info('user')
+        self.assertEqual('enabled', user.status)
+        self.assertEqual('group2', user.member_of[1])
+        self.assertEqual('policyA', user.policies[0])