| # Licensed to the Apache Software Foundation (ASF) under one |
| # or more contributor license agreements. See the NOTICE file |
| # distributed with this work for additional information |
| # regarding copyright ownership. The ASF licenses this file |
| # to you 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. |
| # |
| # this module wraps mc minio client using admin credentials |
| # to perform various operations |
| |
| import logging |
| import json |
| import subprocess |
| import openserverless.config as cfg |
| import openserverless.template as ntp |
| import openserverless.util as util |
| import os |
| |
| class MinioClient: |
| |
| def __init__(self): |
| self.minio_api_host = cfg.get("minio.host", "MINIO_API_HOST", "openserverless-minio") |
| self.minio_api_port = cfg.get("9000", "MINIO_API_PORT", "9000") |
| self.admin_username = cfg.get("minio.admin.user", "MINIO_ADMIN_USER", "minioadmin") |
| self.admin_password = cfg.get("minio.admin.password", "MINIO_ADMIN_PASSWORD", "minioadmin") |
| self.minio_api_url = f"http://{self.minio_api_host}:{self.minio_api_port}" |
| self.alias = "local" |
| |
| # automatically adds the local alias to mc configuration to operate on openserverless main MINIO server |
| self.mc("alias","set", self.alias, self.minio_api_url, self.admin_username, self.admin_password) |
| |
| # execute minio commands using the mc cli tools installed by default inside the operator |
| def mc(self, *kwargs): |
| cmd = ["mc"] |
| cmd += list(kwargs) |
| |
| # executing |
| logging.debug(cmd) |
| try: |
| res = subprocess.run(cmd, capture_output=True) |
| |
| returncode = res.returncode |
| output = res.stdout.decode() |
| error = res.stderr.decode() |
| |
| if returncode != 0: |
| self.last_output = error |
| logging.error(error) |
| else: |
| self.last_output = output |
| logging.info(output) |
| |
| return returncode == 0 |
| except Exception as e: |
| logging.error(e) |
| return e |
| |
| def get_last_output(self): |
| """ |
| returns the last output of the executed command |
| """ |
| return self.last_output if hasattr(self, 'last_output') else None |
| |
| |
| def add_user(self, username, secret_key): |
| """ |
| adds a new minio user to the configured minio instance |
| """ |
| res = util.check(self.mc("admin","user","add", self.alias, username, secret_key),"add_user",True) |
| return util.check(self.init_namespace_alias(username, secret_key),"init_namespace_alias",res) |
| |
| def remove_user(self, username): |
| """ |
| removes a minio user to the configured minio instance |
| """ |
| res = util.check(self.mc("admin","user","remove", self.alias, username),"remove_user",True) |
| return util.check(self.remove_namespace_alias(username),"remove_namespace_alias",res) |
| |
| def make_bucket(self, bucket_name): |
| """ |
| adds a new bucket inside the configured minio instance |
| """ |
| return util.check(self.mc("mb",f"{self.alias}/{bucket_name}"),"make_bucket",True) |
| |
| def force_bucket_remove(self, bucket_name): |
| """ |
| removes unconditionally a bucket |
| """ |
| return util.check(self.mc("rb","--force",f"{self.alias}/{bucket_name}"),"force_bucket_remove",True) |
| |
| def make_public_bucket(self, bucket_name): |
| """ |
| adds a new public bucket to the configured minio instance |
| """ |
| res = util.check(self.make_bucket(bucket_name),"make_bucket",True) |
| return util.check(self.mc("anonymous","-r","set","download",f"{self.alias}/{bucket_name}"),"make_public_bucket",res) |
| |
| def assign_quota_to_bucket(self, bucket_name, quota): |
| """ |
| assign the specified quota on the given bucket |
| """ |
| return util.check(self.mc("quota","set",f"{self.alias}/{bucket_name}","--size", f"{quota}m"),"assign_quota_to_bucket",True) |
| |
| |
| def assign_policy_to_user(self, username, policy): |
| """ |
| assign the specified policy to the given username |
| """ |
| return util.check(self.mc("admin","policy","attach",self.alias,policy,"--user", username),"assign_policy_to_user",True) |
| |
| def add_policy(self, policy, path_to_policy_json): |
| """ |
| add a new policy into minio |
| """ |
| return util.check(self.mc("admin","policy","create",self.alias,policy,path_to_policy_json),"add_policy",True) |
| |
| def remove_policy(self, policy): |
| """ |
| add a new policy into minio |
| """ |
| return util.check(self.mc("admin","policy","remove",self.alias,policy),"remove_policy",True) |
| |
| def render_policy(self,username,template,data): |
| """ |
| uses the given template policy to render a final policy and returns the absolute path to rendered policy file. |
| """ |
| out = f"/tmp/__{username}_{template}" |
| file = ntp.spool_template(template, out, data) |
| return os.path.abspath(file) |
| |
| def assign_rw_bucket_policy_to_user(self,username,bucket_names): |
| """ |
| defines a rw policy template for the specified bucket and assigns it to the given username. |
| """ |
| policy_name = f"{username}_rw_policy" |
| path_to_policy_json = self.render_policy(username,"minio_rw_policy_tpl.json",{"bucket_arns":bucket_names}) |
| res=util.check(self.add_policy(policy_name,path_to_policy_json),"add_policy",True) |
| res=util.check(self.assign_policy_to_user(username,policy_name),"assign_rw_bucket_policy_to_user",res) |
| os.remove(path_to_policy_json) |
| return res |
| |
| def delete_user(self,username): |
| """ |
| removes the user and the corresponding policy |
| """ |
| policy_name = f"{username}_rw_policy" |
| res=util.check(self.remove_user(username),"removed_user",True) |
| return util.check(self.remove_policy(policy_name),"deleted_user_policy",res) |
| |
| def init_namespace_alias(self, namespace, secret_key): |
| """ |
| called to initialize a local minio alias for a namespace user (used to impersonate the user uploading content) |
| """ |
| return util.check(self.mc("alias","set", f"local_{namespace}", self.minio_api_url, namespace, secret_key),"init_namespace_alias_call",True) |
| |
| def remove_namespace_alias(self, namespace): |
| """ |
| called to initialize to remove a local minio alias for a namespace user |
| """ |
| return util.check(self.mc("alias","remove",f"local_{namespace}"),"remove_namespace_alias_call",True) |
| |
| def upload_folder_content(self,origin,bucket): |
| """ |
| uploads the given content using a local alias for the corresponding namespace |
| """ |
| return util.check(self.mc("cp","-r",origin,f"{self.alias}/{bucket}"),"upload_folder_content",True) |
| |
| |
| |
