# 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 nuvolaris.config as cfg import nuvolaris.template as ntp import nuvolaris.util as util import os class MinioClient: def __init__(self): self.minio_api_host = cfg.get("minio.host", "MINIO_API_HOST", "nuvolaris-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 nuvolaris 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: logging.error(error) else: logging.info(output) return returncode == 0 except Exception as e: logging.error(e) return e 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)