import time import numpy as np import pytest from base.client_v2_base import TestMilvusClientV2Base from common import common_func as cf from common import common_type as ct from common.common_type import CaseLabel, CheckTasks from pymilvus import DataType from utils.util_log import test_log as log from utils.util_pymilvus import * # noqa: F403 # RBAC test classes share teardown logic that lists and drops every user, # role, privilege group, and database on the server. Under pytest-xdist `-n` # this cross-deletes objects owned by sibling workers. Pin the whole module # to a single xdist worker (loadgroup) so tests run serially without # starving other test files of parallelism. pytestmark = pytest.mark.xdist_group(name="rbac_serial") prefix = "client_rbac" def _teardown_rbac(test_instance): """Common teardown: drop all non-default users, roles, privilege groups, and databases""" client = test_instance._client() # drop users (revoke roles first) users, _ = test_instance.list_users(client) for user in users: if user != ct.default_user: try: user_info, _ = test_instance.describe_user(client, user) if user_info and user_info.get("roles"): for role in user_info["roles"]: try: test_instance.revoke_role(client, user, role) except Exception: pass test_instance.drop_user(client, user) except Exception: pass # collect all dbs for cross-db privilege revocation dbs, _ = test_instance.list_databases(client) # drop roles (revoke privileges across all dbs first) roles, _ = test_instance.list_roles(client) for role in roles: if role not in ["admin", "public"]: for db in dbs: try: role_info, _ = test_instance.describe_role(client, role, db_name=db) if role_info and role_info.get("privileges"): for priv in role_info["privileges"]: try: test_instance.revoke_privilege( client, role, priv["object_type"], priv["privilege"], priv["object_name"], db_name=priv.get("db_name", ""), ) except Exception: try: test_instance.revoke_privilege_v2( client, role, priv["privilege"], priv.get("object_name", "*"), db_name=priv.get("db_name", "*"), ) except Exception: pass except Exception: pass try: test_instance.drop_role(client, role) except Exception: pass # drop custom privilege groups try: groups, _ = test_instance.list_privilege_groups(client) for g in groups: if g.get("privilege_group") not in ct.built_in_privilege_groups: try: test_instance.drop_privilege_group(client, g["privilege_group"]) except Exception: pass except Exception: pass # drop databases for db in dbs: if db != ct.default_db: try: test_instance.using_database(client, db) colls, _ = test_instance.list_collections(client) for c in colls: test_instance.drop_collection(client, c) test_instance.using_database(client, "default") test_instance.drop_database(client, db) except Exception: try: test_instance.using_database(client, "default") except Exception: pass user_pre = "user" role_pre = "role" root_token = "root:Milvus" default_nb = ct.default_nb default_nq = ct.default_nq default_dim = ct.default_dim default_limit = ct.default_limit default_search_exp = "id >= 0" exp_res = "exp_res" default_search_field = ct.default_float_vec_field_name default_search_params = ct.default_search_params default_primary_key_field_name = "id" default_vector_field_name = "vector" default_float_field_name = ct.default_float_field_name default_string_field_name = ct.default_string_field_name @pytest.mark.tags(CaseLabel.RBAC) class TestMilvusClientRbacBase(TestMilvusClientV2Base): """Test case of rbac interface""" def teardown_method(self, method): log.info("[teardown_method] Start teardown RBAC test cases ...") _teardown_rbac(self) super().teardown_method(method) def test_milvus_client_connect_using_token(self, host, port): """ target: test init milvus client using token method: init milvus client with only token expected: init successfully """ uri = f"http://{host}:{port}" client, _ = self.init_milvus_client(uri=uri, token=root_token) # check success link res = self.list_databases(client)[0] assert res != [] def test_milvus_client_connect_using_user_password(self, host, port): """ target: test init milvus client using user and password method: init milvus client with user and password expected: init successfully """ uri = f"http://{host}:{port}" client, _ = self.init_milvus_client(uri=uri, user=ct.default_user, password=ct.default_password) # check success link res = self.list_databases(client)[0] assert res != [] def test_milvus_client_create_user(self, host, port): """ target: test milvus client api create_user method: create user expected: succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length() self.create_user(client, user_name=user_name, password=password) # check uri = f"http://{host}:{port}" client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) res = self.list_databases(client)[0] assert res == [] def test_milvus_client_drop_user(self, host, port): """ target: test milvus client api drop_user method: drop user expected: succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length() self.create_user(client, user_name=user_name, password=password) # drop user that exists self.drop_user(client, user_name=user_name) # drop user that not exists not_exist_user_name = cf.gen_unique_str(user_pre) self.drop_user(client, user_name=not_exist_user_name) def test_milvus_client_update_password(self, host, port): """ target: test milvus client api update_password method: create a user and update password expected: succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length() self.create_user(client, user_name=user_name, password=password) new_password = cf.gen_str_by_length() self.update_password(client, user_name=user_name, old_password=password, new_password=new_password) self.close(client) # check uri = f"http://{host}:{port}" client, _ = self.init_milvus_client(uri=uri, user=user_name, password=new_password) res = self.list_databases(client)[0] assert res == [] self.close(client) self.init_milvus_client(uri=uri, user=user_name, password=password, check_task=CheckTasks.check_auth_failure) def test_milvus_client_list_users(self, host, port): """ target: test milvus client api list_users method: create a user and list users expected: succeed """ client = self._client() user_name1 = cf.gen_unique_str(user_pre) user_name2 = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length() self.create_user(client, user_name=user_name1, password=password) self.create_user(client, user_name=user_name2, password=password) res = self.list_users(client)[0] assert {ct.default_user, user_name1, user_name2}.issubset(set(res)) is True def test_milvus_client_describe_user(self, host, port): """ target: test milvus client api describe_user method: create a user and describe the user expected: succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length() self.create_user(client, user_name=user_name, password=password) # describe one self res, _ = self.describe_user(client, user_name=ct.default_user) assert res["user_name"] == ct.default_user # describe other users res, _ = self.describe_user(client, user_name=user_name) assert res["user_name"] == user_name # describe user that not exists user_not_exist = cf.gen_unique_str(user_pre) res, _ = self.describe_user(client, user_name=user_not_exist) assert res == {} def test_milvus_client_create_role(self, host, port): """ target: test milvus client api create_role method: create a role expected: succeed """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) def test_milvus_client_drop_role(self, host, port): """ target: test milvus client api drop_role method: create a role and drop expected: succeed """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) self.drop_role(client, role_name=role_name) def test_milvus_client_describe_role(self, host, port): """ target: test milvus client api describe_role method: create a role and describe expected: succeed """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) # describe a role that exists self.describe_role(client, role_name=role_name) def test_milvus_client_list_roles(self, host, port): """ target: test milvus client api list_roles method: create a role and list roles expected: succeed """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) res, _ = self.list_roles(client) assert role_name in res def test_milvus_client_grant_role(self, host, port): """ target: test milvus client api grant_role method: create a role and a user, then grant role to the user expected: succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) role_name = cf.gen_unique_str(role_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) def test_milvus_client_revoke_role(self, host, port): """ target: test milvus client api revoke_role method: create a role and a user, grant role + Search privilege, verify search succeeds, then revoke role and verify search is denied expected: succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) role_name = cf.gen_unique_str(role_pre) password = cf.gen_str_by_length(contain_numbers=True) collection_name = cf.gen_unique_str(prefix) # create user and role self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) # create collection, insert data, create index and load self.create_collection(client, collection_name, default_dim, consistency_level="Strong") rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(client, collection_name, rows) index_params = self.prepare_index_params(client)[0] index_params.add_index(default_vector_field_name, metric_type="COSINE") self.create_index(client, collection_name, index_params) self.load_collection(client, collection_name) # grant role + Search privilege self.grant_role(client, user_name=user_name, role_name=role_name) self.grant_privilege(client, role_name, "Collection", "Search", collection_name) time.sleep(10) # verify search succeeds with the role uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) vectors_to_search = cf.gen_vectors(1, default_dim) self.search(user_client, collection_name, vectors_to_search) # revoke the role from user self.revoke_role(client, user_name=user_name, role_name=role_name) time.sleep(10) # verify search is denied after revoking role self.search(user_client, collection_name, vectors_to_search, check_task=CheckTasks.check_permission_deny) # cleanup self.drop_collection(client, collection_name) def test_milvus_client_grant_privilege(self, host, port): """ target: test milvus client api grant_privilege method: create a role and a user, then grant role to the user, grant a privilege to the role expected: succeed """ # prepare a collection client_root = self._client() coll_name = cf.gen_unique_str() self.create_collection(client_root, coll_name, default_dim, consistency_level="Strong") # create a new role and a new user ( no privilege) user_name = cf.gen_unique_str(user_pre) role_name = cf.gen_unique_str(role_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client_root, user_name=user_name, password=password) self.create_role(client_root, role_name=role_name) self.grant_role(client_root, user_name=user_name, role_name=role_name) # check the role has no privilege of drop collection uri = f"http://{host}:{port}" client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) self.drop_collection(client, coll_name, check_task=CheckTasks.check_permission_deny) # grant the role with the privilege of drop collection self.grant_privilege(client_root, role_name, "Global", "DropCollection", "*") time.sleep(10) # check the role has privilege of drop collection self.drop_collection(client, coll_name) def test_milvus_client_revoke_privilege(self, host, port): """ target: test milvus client api revoke_privilege method: create a role and a user, then grant role to the user, grant a privilege to the role, then revoke expected: succeed """ # prepare a collection client_root = self._client() coll_name = cf.gen_unique_str() # create a new role and a new user ( no privilege) user_name = cf.gen_unique_str(user_pre) role_name = cf.gen_unique_str(role_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client_root, user_name=user_name, password=password) self.create_role(client_root, role_name=role_name) self.grant_role(client_root, user_name=user_name, role_name=role_name) # FIX: CreateCollection is a cluster-level privilege, needs db_name="*" self.grant_privilege_v2(client_root, role_name, "CreateCollection", collection_name="*", db_name="*") time.sleep(20) # check the role has privilege of create collection uri = f"http://{host}:{port}" client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) # Use schema mode to avoid auto-index/load which needs extra privileges schema = self.create_schema(client)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True) schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) self.create_collection(client, coll_name, schema=schema) # revoke the role with the privilege of create collection self.revoke_privilege_v2(client_root, role_name, "CreateCollection", collection_name="*", db_name="*") time.sleep(20) # reconnect to pick up revoked privilege client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) # check the role has no privilege of create collection coll_name_2 = cf.gen_unique_str() schema2 = self.create_schema(client)[0] schema2.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True) schema2.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) self.create_collection(client, coll_name_2, schema=schema2, check_task=CheckTasks.check_permission_deny) @pytest.mark.tags(CaseLabel.RBAC) class TestMilvusClientRbacInvalid(TestMilvusClientV2Base): """Test case of rbac interface""" def teardown_method(self, method): log.info("[teardown_method] Start teardown RBAC invalid test cases ...") _teardown_rbac(self) super().teardown_method(method) def test_milvus_client_init_token_invalid(self, host, port): """ target: test milvus client api token invalid method: init milvus client using a wrong token expected: raise exception """ uri = f"http://{host}:{port}" wrong_token = root_token + "kk" self.init_milvus_client(uri=uri, token=wrong_token, check_task=CheckTasks.check_auth_failure) def test_milvus_client_init_username_invalid(self, host, port): """ target: test milvus client api username invalid method: init milvus client using a wrong username expected: raise exception """ uri = f"http://{host}:{port}" invalid_user_name = ct.default_user + "nn" self.init_milvus_client( uri=uri, user=invalid_user_name, password=ct.default_password, check_task=CheckTasks.check_auth_failure ) def test_milvus_client_init_password_invalid(self, host, port): """ target: test milvus client api password invalid method: init milvus client using a wrong password expected: raise exception """ uri = f"http://{host}:{port}" wrong_password = ct.default_password + "kk" self.init_milvus_client( uri=uri, user=ct.default_user, password=wrong_password, check_task=CheckTasks.check_auth_failure ) @pytest.mark.parametrize("invalid_name", ["", "0", "n@me", "h h"]) def test_milvus_client_create_user_value_invalid(self, host, port, invalid_name): """ target: test milvus client api create_user invalid method: create using a wrong username expected: raise exception """ client = self._client() self.create_user( client, invalid_name, ct.default_password, check_task=CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "invalid parameter"}, ) @pytest.mark.parametrize("invalid_name", [1, [], None, {}]) def test_milvus_client_create_user_type_invalid(self, host, port, invalid_name): """ target: test milvus client api create_user invalid method: create using a wrong username expected: raise exception """ client = self._client() self.create_user( client, invalid_name, ct.default_password, check_task=CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: f"`user` value {invalid_name} is illegal"}, ) def test_milvus_client_create_user_exist(self, host, port): """ target: test milvus client api create_user invalid method: create using a wrong username expected: raise exception """ client = self._client() self.create_user( client, "root", ct.default_password, check_task=CheckTasks.err_res, check_items={ct.err_code: 65535, ct.err_msg: "user already exists: root"}, ) @pytest.mark.parametrize("invalid_password", ["", "0", "p@ss", "h h", "1+1=2"]) def test_milvus_client_create_user_password_invalid_value(self, host, port, invalid_password): """ target: test milvus client api create_user invalid method: create using a wrong username expected: raise exception """ client = self._client() user_name = cf.gen_unique_str(user_pre) self.create_user( client, user_name, invalid_password, check_task=CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "invalid password"}, ) @pytest.mark.parametrize("invalid_password", [1, [], None, {}]) def test_milvus_client_create_user_password_invalid_type(self, host, port, invalid_password): """ target: test milvus client api create_user invalid method: create using a wrong username expected: raise exception """ client = self._client() user_name = cf.gen_unique_str(user_pre) self.create_user( client, user_name, invalid_password, check_task=CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: f"`password` value {invalid_password} is illegal"}, ) def test_milvus_client_update_password_user_not_exist(self, host, port): """ target: test milvus client api update_password method: create a user and update password expected: raise exception """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) new_password = cf.gen_str_by_length(contain_numbers=True) self.update_password( client, user_name=user_name, old_password=password, new_password=new_password, check_task=CheckTasks.err_res, check_items={ct.err_code: 1400, ct.err_msg: f"old password not correct for {user_name}: not authenticated"}, ) def test_milvus_client_update_password_password_wrong(self, host, port): """ target: test milvus client api update_password method: create a user and update password expected: succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=user_name, password=password) new_password = cf.gen_str_by_length(contain_numbers=True) wrong_password = password + "kk" self.update_password( client, user_name=user_name, old_password=wrong_password, new_password=new_password, check_task=CheckTasks.err_res, check_items={ct.err_code: 1400, ct.err_msg: f"old password not correct for {user_name}: not authenticated"}, ) def test_milvus_client_update_password_new_password_same(self, host, port): """ target: test milvus client api update_password method: create a user and update password expected: succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=user_name, password=password) self.update_password(client, user_name=user_name, old_password=password, new_password=password) @pytest.mark.parametrize("invalid_password", ["", "0", "p@ss", "h h", "1+1=2"]) def test_milvus_client_update_password_new_password_invalid(self, host, port, invalid_password): """ target: test milvus client api update_password method: create a user and update password expected: succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=user_name, password=password) self.update_password( client, user_name=user_name, old_password=password, new_password=invalid_password, check_task=CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: "invalid password"}, ) def test_milvus_client_create_role_exist(self, host, port): """ target: test milvus client api create_role method: create a role using invalid name expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) # create existed role error_msg = f'role [name:"{role_name}"] already exists' self.create_role( client, role_name=role_name, check_task=CheckTasks.err_res, check_items={ct.err_code: 65535, ct.err_msg: error_msg}, ) def test_milvus_client_drop_role_invalid(self, host, port): """ target: test milvus client api drop_role method: create a role and drop expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.drop_role( client, role_name=role_name, check_task=CheckTasks.err_res, check_items={ ct.err_code: 65535, ct.err_msg: "not found the role, maybe the role isn't existed or internal system error", }, ) @pytest.mark.parametrize("role_name", ["admin", "public"]) def test_milvus_client_drop_built_in_role(self, host, port, role_name): """ target: test milvus client api drop_role method: create a role and drop expected: raise exception """ client = self._client() self.drop_role( client, role_name=role_name, check_task=CheckTasks.err_res, check_items={ ct.err_code: 65535, ct.err_msg: f"the role[{role_name}] is a default role, which can't be dropped", }, ) def test_milvus_client_describe_role_invalid(self, host, port): """ target: test milvus client api describe_role method: describe a role using invalid name expected: raise exception """ client = self._client() # describe a role that does not exist role_not_exist = cf.gen_unique_str(role_pre) error_msg = "not found the role, maybe the role isn't existed or internal system error" self.describe_role( client, role_name=role_not_exist, check_task=CheckTasks.err_res, check_items={ct.err_code: 65535, ct.err_msg: error_msg}, ) def test_milvus_client_grant_role_user_not_exist(self, host, port): """ target: test milvus client api grant_role method: create a role and a user, then grant role to the user expected: succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) self.grant_role( client, user_name=user_name, role_name=role_name, check_task=CheckTasks.err_res, check_items={ ct.err_code: 65536, ct.err_msg: "not found the user, maybe the user isn't existed or internal system error", }, ) def test_milvus_client_grant_role_role_not_exist(self, host, port): """ target: test milvus client api grant_role method: create a role and a user, then grant role to the user expected: succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) role_name = cf.gen_unique_str(role_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=user_name, password=password) self.grant_role( client, user_name=user_name, role_name=role_name, check_task=CheckTasks.err_res, check_items={ ct.err_code: 65536, ct.err_msg: "not found the role, maybe the role isn't existed or internal system error", }, ) @pytest.mark.parametrize("name", [1, 1.0]) def test_milvus_client_create_privilege_group_with_privilege_group_name_invalid_type(self, name, host, port): """ target: test milvus client api create privilege group with invalid name method: create privilege group with invalid name expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) error_msg = f"`privilege_group` value {name} is illegal" self.create_privilege_group( client, privilege_group=name, check_task=CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: error_msg}, ) # cleanup self.drop_role(client, role_name=role_name) @pytest.mark.parametrize("name", ["n%$#@!", "test-role", "ff ff"]) def test_milvus_client_create_privilege_group_with_privilege_group_name_invalid_value_1(self, name, host, port): """ target: test milvus client api create privilege group with invalid name method: create privilege group with invalid name expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) error_msg = f"privilege group name {name} can only contain numbers, letters and underscores: invalid parameter" self.create_privilege_group( client, privilege_group=name, check_task=CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: error_msg}, ) # cleanup self.drop_role(client, role_name=role_name) @pytest.mark.parametrize("name", [1, 1.0]) def test_milvus_client_drop_privilege_group_with_privilege_group_name_invalid_type(self, name, host, port): """ target: test milvus client api drop privilege group with invalid name method: drop privilege group with invalid name expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) error_msg = f"`privilege_group` value {name} is illegal" self.drop_privilege_group( client, privilege_group=name, check_task=CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: error_msg}, ) # cleanup self.drop_role(client, role_name=role_name) @pytest.mark.parametrize("name", [1, 1.0]) def test_milvus_client_add_privileges_to_group_with_privilege_group_name_invalid_type(self, name, host, port): """ target: test milvus client api add privilege group with invalid name method: add privilege group with invalid name expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) error_msg = f"`privilege_group` value {name} is illegal" self.add_privileges_to_group( client, privilege_group=name, privileges=["Insert"], check_task=CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: error_msg}, ) # cleanup self.drop_role(client, role_name=role_name) @pytest.mark.parametrize("name", ["n%$#@!", "test-role", "ff ff"]) def test_milvus_client_add_privileges_to_group_with_privilege_group_name_invalid_value(self, name, host, port): """ target: test milvus client api add privilege group with invalid name method: add privilege group with invalid name expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) error_msg = f"privilege group name {name} can only contain numbers, letters and underscores: invalid parameter" self.add_privileges_to_group( client, privilege_group=name, privileges=["Insert"], check_task=CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: error_msg}, ) # cleanup self.drop_role(client, role_name=role_name) def test_milvus_client_add_privilege_into_not_exist_privilege_group(self, host, port): """ target: test milvus client api add privilege into not exist privilege group method: add privilege into not exist privilege group expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) privilege_group_name = "privilege_group_not_exist" error_msg = f"there is no privilege group name [{privilege_group_name}] defined in system" self.add_privileges_to_group( client, privilege_group=privilege_group_name, privileges=["Insert"], check_task=CheckTasks.err_res, check_items={ct.err_code: 1100, ct.err_msg: error_msg}, ) # cleanup self.drop_role(client, role_name=role_name) @pytest.mark.parametrize("name", [1, 1.0, "n%$#@!", "test-role", "ff ff", "invalid_privilege"]) def test_milvus_client_add_privileges_to_group_with_privilege_invalid(self, name, host, port): """ target: test milvus client api add privilege group with invalid privilege type method: add privilege group with invalid privilege type expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) error_msg = f"`privileges` value {name} is illegal" self.add_privileges_to_group( client, privilege_group="privilege_group_1", privileges=name, check_task=CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: error_msg}, ) # cleanup self.drop_role(client, role_name=role_name) @pytest.mark.parametrize("name", [1, 1.0, "n%$#@!", "test-role", "ff ff"]) def test_milvus_client_remove_privileges_to_group_with_privilege_group_name_invalid(self, name, host, port): """ target: test milvus client api remove privilege group with invalid name method: remove privilege group with invalid name expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) error_msg = f"{name}" self.remove_privileges_from_group( client, privilege_group=name, privileges=["Insert"], check_task=CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: error_msg}, ) # cleanup self.drop_role(client, role_name=role_name) @pytest.mark.parametrize("name", [1, 1.0, "n%$#@!", "test-role", "ff ff", "invalid_privilege"]) def test_milvus_client_remove_privileges_to_group_with_privilege_invalid(self, name, host, port): """ target: test milvus client api remove privilege group with invalid privilege type method: remove privilege group with invalid privilege type expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) error_msg = f"`privileges` value {name} is illegal" self.remove_privileges_from_group( client, privilege_group="privilege_group_1", privileges=name, check_task=CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: error_msg}, ) # cleanup self.drop_role(client, role_name=role_name) # ==================== P2 Edge Cases ==================== def test_milvus_client_remove_root_from_default_role(self, host, port): """ target: test milvus client api revoke root from admin and public role method: try to revoke root user from admin and public role, then restore expected: revoke succeeds (no error), root is restored afterwards """ client = self._client() try: # revoke root from admin role — succeeds without error self.revoke_role(client, user_name=ct.default_user, role_name="admin") # revoke root from public role — succeeds without error self.revoke_role(client, user_name=ct.default_user, role_name="public") finally: # FIX: restore root's admin and public roles after revoke self.grant_role(client, user_name=ct.default_user, role_name="admin") self.grant_role(client, user_name=ct.default_user, role_name="public") # verify root has admin role restored user_info, _ = self.describe_user(client, user_name=ct.default_user) assert "admin" in user_info.get("roles", []) def test_milvus_client_remove_user_from_unbind_role(self, host, port): """ target: test milvus client api revoke role from user who is not bound method: create user and role (don't bind), then revoke expected: server silently succeeds """ client = self._client() user_name = cf.gen_unique_str(user_pre) role_name = cf.gen_unique_str(role_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) # LOGIC CHANGE: server now silently succeeds on revoke_role for unbound user self.revoke_role(client, user_name=user_name, role_name=role_name) def test_milvus_client_remove_user_from_nonexistent_role(self, host, port): """ target: test milvus client api revoke user from nonexistent role method: create user, try to revoke nonexistent role expected: raise error because the role does not exist """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=user_name, password=password) nonexistent_role = cf.gen_unique_str(role_pre) # FIX: revoking from a nonexistent role returns an error self.revoke_role( client, user_name=user_name, role_name=nonexistent_role, check_task=CheckTasks.err_res, check_items={ct.err_code: 65535, ct.err_msg: "not found the role"}, ) def test_milvus_client_remove_nonexistent_user_from_role(self, host, port): """ target: test milvus client api revoke nonexistent user from role method: create role, try to revoke nonexistent user expected: server silently succeeds """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) nonexistent_user = cf.gen_unique_str(user_pre) # LOGIC CHANGE: server now silently succeeds on revoke_role for nonexistent user self.revoke_role(client, user_name=nonexistent_user, role_name=role_name) @pytest.mark.parametrize("role_name", ["admin", "public"]) def test_milvus_client_remove_user_from_default_role(self, role_name, host, port): """ target: test milvus client api revoke user from default role after granting method: create user, grant admin/public, revoke expected: succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=user_name, password=password) self.grant_role(client, user_name=user_name, role_name=role_name) # verify user has the role user_info, _ = self.describe_user(client, user_name=user_name) assert role_name in user_info.get("roles", []) # revoke role from user self.revoke_role(client, user_name=user_name, role_name=role_name) # verify user no longer has the role user_info, _ = self.describe_user(client, user_name=user_name) assert role_name not in user_info.get("roles", []) def test_milvus_client_remove_root_from_new_role(self, host, port): """ target: test milvus client api grant and revoke root from new role method: create role, grant to root, verify, revoke, verify expected: succeed """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) # grant role to root self.grant_role(client, user_name=ct.default_user, role_name=role_name) # verify root has the role user_info, _ = self.describe_user(client, user_name=ct.default_user) assert role_name in user_info.get("roles", []) # revoke role from root self.revoke_role(client, user_name=ct.default_user, role_name=role_name) # verify root no longer has the role user_info, _ = self.describe_user(client, user_name=ct.default_user) assert role_name not in user_info.get("roles", []) def test_milvus_client_list_grant_by_role_and_not_exist_object(self, host, port): """ target: test milvus client api describe role with no grants method: create role, describe role expected: privileges list is empty """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) role_info, _ = self.describe_role(client, role_name=role_name) assert role_info.get("privileges", []) == [] def test_milvus_client_revoke_privilege_with_object_not_exist(self, host, port): """ target: test milvus client api revoke privilege with random object type method: create role, revoke privilege with random object type expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) random_object = cf.gen_unique_str(prefix) self.revoke_privilege( client, role_name, random_object, "*", "*", check_task=CheckTasks.err_res, check_items={ct.err_code: 65535, ct.err_msg: "the object entity in the request is nil or invalid"}, ) def test_milvus_client_revoke_privilege_with_privilege_not_exist(self, host, port): """ target: test milvus client api revoke privilege with random privilege name method: create role, revoke privilege with random privilege name expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) random_privilege = cf.gen_unique_str(prefix) self.revoke_privilege( client, role_name, "Global", random_privilege, "*", check_task=CheckTasks.err_res, check_items={ct.err_code: 65535, ct.err_msg: "not found the privilege name"}, ) def test_milvus_client_grant_privilege_with_db_not_exist(self, host, port): """ target: test milvus client api grant privilege with nonexistent db method: create user/role, grant on nonexistent db, root client using_database expected: raise exception when using nonexistent db """ client = self._client() user_name = cf.gen_unique_str(user_pre) role_name = cf.gen_unique_str(role_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) nonexistent_db = cf.gen_unique_str(prefix) self.grant_privilege(client, role_name, "Global", "All", "*", db_name=nonexistent_db) time.sleep(10) self.using_database( client, nonexistent_db, check_task=CheckTasks.err_res, check_items={ct.err_code: 2, ct.err_msg: "database not found"}, ) def test_milvus_client_revoke_db_not_existed(self, host, port): """ target: test milvus client api revoke privilege with nonexistent db method: grant Global All, revoke on nonexistent db expected: silent success """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) self.grant_privilege(client, role_name, "Global", "All", "*") nonexistent_db = cf.gen_unique_str(prefix) self.revoke_privilege(client, role_name, "Global", "All", "*", db_name=nonexistent_db) def test_milvus_client_list_grant_db_non_existed(self, host, port): """ target: test milvus client api describe role with nonexistent db method: grant Global All, describe role with nonexistent db expected: privileges list is empty """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) self.grant_privilege(client, role_name, "Global", "All", "*") nonexistent_db = cf.gen_unique_str(prefix) role_info, _ = self.describe_role(client, role_name=role_name, db_name=nonexistent_db) assert role_info.get("privileges", []) == [] def test_milvus_client_grant_v2_collection_name_invalid_type(self, host, port): """ target: test milvus client api grant_privilege_v2 with invalid collection_name type method: grant_privilege_v2 with collection_name=1 expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) self.grant_privilege_v2( client, role_name, "Insert", collection_name=1, check_task=CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: "is illegal"}, ) def test_milvus_client_grant_v2_db_name_invalid_type(self, host, port): """ target: test milvus client api grant_privilege_v2 with invalid db_name type method: grant_privilege_v2 with db_name=1 expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) self.grant_privilege_v2( client, role_name, "Insert", collection_name="*", db_name=1, check_task=CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: "is illegal"}, ) def test_milvus_client_grant_v2_db_name_invalid_value(self, host, port): """ target: test milvus client api grant_privilege_v2 with invalid db_name value method: grant_privilege_v2 with db_name="n%$#@!" expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) self.grant_privilege_v2( client, role_name, "Insert", collection_name="*", db_name="n%$#@!", check_task=CheckTasks.err_res, check_items={ct.err_code: 802, ct.err_msg: "database name can only contain"}, ) def test_milvus_client_grant_v2_not_exist_db_name(self, host, port): """ target: test milvus client api grant_privilege_v2 with nonexistent db_name method: create privilege group, grant_v2 with nonexistent db, revoke_v2 expected: succeed (no error) """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) pg_name = cf.gen_unique_str(prefix) self.create_privilege_group(client, privilege_group=pg_name) self.add_privileges_to_group(client, privilege_group=pg_name, privileges=["Insert"]) nonexistent_db = cf.gen_unique_str(prefix) self.grant_privilege_v2(client, role_name, pg_name, collection_name="*", db_name=nonexistent_db) self.revoke_privilege_v2(client, role_name, pg_name, collection_name="*", db_name=nonexistent_db) # cleanup self.remove_privileges_from_group(client, privilege_group=pg_name, privileges=["Insert"]) self.drop_privilege_group(client, privilege_group=pg_name) def test_milvus_client_revoke_v2_privilege_invalid_type(self, host, port): """ target: test milvus client api revoke_privilege_v2 with invalid privilege type method: revoke_privilege_v2 with privilege=1 expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) self.revoke_privilege_v2( client, role_name, privilege=1, collection_name="*", check_task=CheckTasks.err_res, check_items={ct.err_code: 1, ct.err_msg: "is illegal"}, ) def test_milvus_client_revoke_v2_privilege_invalid_value(self, host, port): """ target: test milvus client api revoke_privilege_v2 with invalid privilege value method: revoke_privilege_v2 with random privilege name expected: raise exception """ client = self._client() role_name = cf.gen_unique_str(role_pre) self.create_role(client, role_name=role_name) random_privilege = cf.gen_unique_str(prefix) self.revoke_privilege_v2( client, role_name, privilege=random_privilege, collection_name="*", check_task=CheckTasks.err_res, check_items={ct.err_code: 65535, ct.err_msg: "not found the privilege name"}, ) def test_milvus_client_delete_all_users(self, host, port): """ target: test milvus client api delete all non-root users method: create 3 users, delete all non-root users expected: only root user remains """ client = self._client() user_names = [] for _ in range(3): user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=user_name, password=password) user_names.append(user_name) # verify all users exist users, _ = self.list_users(client) for un in user_names: assert un in users # delete all non-root users for un in user_names: self.drop_user(client, user_name=un) # verify only root remains users, _ = self.list_users(client) assert len(users) == 1 assert ct.default_user in users @pytest.mark.tags(CaseLabel.RBAC) class TestMilvusClientRbacAdvance(TestMilvusClientV2Base): """Test case of rbac interface""" def teardown_method(self, method): log.info("[teardown_method] Start teardown RBAC test cases ...") _teardown_rbac(self) super().teardown_method(method) # NOTE: requires common.security.authorizationEnabled=true @pytest.mark.skip("requires specific multi-db RBAC setup") def test_milvus_client_search_iterator_rbac_mul_db(self): """ target: test search iterator(high level api) normal case about mul db by rbac method: create connection, collection, insert and search iterator expected: search iterator permission deny after switch to no permission db """ batch_size = 20 uri = f"http://{cf.param_info.param_host}:{cf.param_info.param_port}" client, _ = self.init_milvus_client(uri=uri, token="root:Milvus") my_db = cf.gen_unique_str(prefix) self.create_database(client, my_db) collection_name = cf.gen_unique_str(prefix) self.using_database(client, my_db) # 1. create collection self.create_collection(client, collection_name, default_dim, consistency_level="Bounded") # 2. insert rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(client, collection_name, rows) self.flush(client, collection_name) self.using_database(client, "default") # 3. create collection self.create_collection(client, collection_name, default_dim, consistency_level="Bounded") # 4. insert self.insert(client, collection_name, rows) self.flush(client, collection_name) user_name = cf.gen_unique_str(user_pre) role_name = cf.gen_unique_str(role_pre) password = cf.gen_str_by_length() self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.grant_privilege(client, role_name, "Collection", "Search", collection_name, "default") self.grant_privilege(client, role_name, "Collection", "Insert", collection_name, my_db) client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) # 5. search_iterator vectors_to_search = rng.random((1, default_dim)) self.search_iterator( client, collection_name, vectors_to_search, batch_size, use_rbac_mul_db=True, another_db=my_db, check_task=CheckTasks.check_permission_deny, ) client, _ = self.init_milvus_client(uri=uri, token="root:Milvus") self.revoke_privilege(client, role_name, "Collection", "Insert", collection_name, my_db) self.drop_collection(client, collection_name) self.using_database(client, "default") self.drop_collection(client, collection_name) def test_milvus_client_drop_role_which_bind_user(self, host, port): """ target: test milvus client api drop_role when role is bound to user method: create a role, bind user to the role, drop the role expected: drop success """ client = self._client() user_name = cf.gen_unique_str(user_pre) role_name = cf.gen_unique_str(role_pre) password = cf.gen_str_by_length(contain_numbers=True) # create user and role self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) # bind user to role self.grant_role(client, user_name=user_name, role_name=role_name) # drop role that has user bound to it self.drop_role(client, role_name=role_name) # verify role is dropped roles, _ = self.list_roles(client) assert role_name not in roles @pytest.mark.parametrize("role_name", ["admin", "public"]) def test_milvus_client_add_user_to_default_role(self, role_name, host, port): """ target: test milvus client api add user to default role (admin or public) method: create a user, add user to admin role or public role expected: add success """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) # create user self.create_user(client, user_name=user_name, password=password) # grant role to user (add user to role) self.grant_role(client, user_name=user_name, role_name=role_name) # grant role to user again to test idempotency self.grant_role(client, user_name=user_name, role_name=role_name) # verify user has the role user_info, _ = self.describe_user(client, user_name=user_name) assert user_info["user_name"] == user_name assert role_name in user_info.get("roles", []) def test_milvus_client_add_root_to_new_role(self, host, port): """ target: test milvus client api add root to new role method: create a new role and add root user to the role expected: add success """ client = self._client() role_name = cf.gen_unique_str(role_pre) # create new role self.create_role(client, role_name=role_name) # add root user to the new role self.grant_role(client, user_name=ct.default_user, role_name=role_name) # verify root user has the role user_info, _ = self.describe_user(client, user_name=ct.default_user) assert user_info["user_name"] == ct.default_user assert role_name in user_info.get("roles", []) # drop the role self.drop_role(client, role_name=role_name) def test_milvus_client_list_collection_grants_by_role_and_object(self, host, port): """ target: test milvus client api list grants by role and object method: create a new role, grant role collection privilege, list grants by role and object expected: list success """ client = self._client() role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) # create collection self.create_collection(client, collection_name, default_dim, consistency_level="Strong") # create new role self.create_role(client, role_name=role_name) # grant collection privileges to role self.grant_privilege(client, role_name, "Collection", "Search", collection_name) self.grant_privilege(client, role_name, "Collection", "Insert", collection_name) # wait for privilege to take effect time.sleep(10) # describe role to get privileges role_info, _ = self.describe_role(client, role_name=role_name) # verify grants privileges = role_info.get("privileges", []) assert len(privileges) == 2 privilege_names = [priv["privilege"] for priv in privileges] assert "Search" in privilege_names assert "Insert" in privilege_names # verify object type and name for priv in privileges: assert priv["object_type"] == "Collection" assert priv["object_name"] == collection_name # revoke privileges for priv in privileges: self.revoke_privilege(client, role_name, priv["object_type"], priv["privilege"], priv["object_name"]) # drop role and collection self.drop_role(client, role_name=role_name) self.drop_collection(client, collection_name) def test_milvus_client_list_global_grants_by_role_and_object(self, host, port): """ target: test milvus client api list grants by role and object for global privileges method: create a new role, grant role global privilege, list grants by role and object expected: list success """ client = self._client() role_name = cf.gen_unique_str(role_pre) # create new role self.create_role(client, role_name=role_name) # grant global privileges to role self.grant_privilege(client, role_name, "Global", "CreateCollection", "*") self.grant_privilege(client, role_name, "Global", "All", "*") # wait for privilege to take effect time.sleep(10) # describe role to get privileges role_info, _ = self.describe_role(client, role_name=role_name) # verify grants privileges = role_info.get("privileges", []) assert len(privileges) == 2 privilege_names = [priv["privilege"] for priv in privileges] assert "CreateCollection" in privilege_names assert "All" in privilege_names # verify object type and name for priv in privileges: assert priv["object_type"] == "Global" assert priv["object_name"] == "*" # revoke privileges for priv in privileges: self.revoke_privilege(client, role_name, priv["object_type"], priv["privilege"], priv["object_name"]) # wait for revoke to take effect time.sleep(10) # drop role self.drop_role(client, role_name=role_name) def test_milvus_client_verify_admin_role_privilege(self, host, port): """ target: test milvus client api verify admin role privilege method: create a new user, bind to admin role, crud collection expected: verify success """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) collection_name = cf.gen_unique_str(prefix) # create user self.create_user(client, user_name=user_name, password=password) # grant admin role to user self.grant_role(client, user_name=user_name, role_name="admin") # wait for role to take effect time.sleep(10) # create new client with the user credentials uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) # test admin privileges by performing CRUD operations # create collection self.create_collection(user_client, collection_name, default_dim, consistency_level="Strong") # insert data rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(user_client, collection_name, rows) # create index index_params, _ = self.prepare_index_params(user_client) index_params.add_index(field_name=default_vector_field_name) self.create_index(user_client, collection_name, index_params) # load collection self.load_collection(user_client, collection_name) # verify collection has data self.flush(user_client, collection_name) num_entities, _ = self.get_collection_stats(user_client, collection_name) assert num_entities["row_count"] == default_nb # release collection self.release_collection(user_client, collection_name) # drop collection self.drop_collection(user_client, collection_name) # drop user self.drop_user(client, user_name=user_name) @pytest.mark.parametrize("with_db", [False, True]) def test_milvus_client_verify_grant_collection_load_privilege(self, host, port, with_db): """ target: test milvus client api verify grant collection load privilege method: verify grant collection load privilege expected: verify success """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) collection_name = cf.gen_unique_str(prefix) role_name = cf.gen_unique_str(role_pre) # create user and role self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) # handle database parameter if with_db: db_name = cf.gen_unique_str(prefix) self.create_database(client, db_name) self.using_database(client, db_name) else: db_name = "default" # create collection self.create_collection(client, collection_name, default_dim, consistency_level="Strong") # grant collection load privileges to role self.grant_privilege(client, role_name, "Collection", "Load", collection_name, db_name=db_name) self.grant_privilege(client, role_name, "Collection", "GetLoadingProgress", collection_name, db_name=db_name) # wait for privilege to take effect time.sleep(10) # create new client with the user credentials uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password, db_name=db_name) # test load privilege self.load_collection(user_client, collection_name) # cleanup self.drop_collection(client, collection_name) if with_db: self.using_database(client, "default") self.drop_database(client, db_name) @pytest.mark.parametrize("with_db", [False, True]) def test_milvus_client_verify_grant_collection_release_privilege(self, host, port, with_db): """ target: test milvus client api verify grant collection release privilege method: verify grant collection release privilege expected: verify success """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) collection_name = cf.gen_unique_str(prefix) role_name = cf.gen_unique_str(role_pre) # create user and role self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) # handle database parameter if with_db: db_name = cf.gen_unique_str(prefix) self.create_database(client, db_name) self.using_database(client, db_name) else: db_name = "default" # create collection self.create_collection(client, collection_name, default_dim, consistency_level="Strong") # load collection self.load_collection(client, collection_name) # grant collection release privilege to role self.grant_privilege(client, role_name, "Collection", "Release", collection_name, db_name=db_name) # wait for privilege to take effect time.sleep(10) # create new client with the user credentials uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password, db_name=db_name) # test release privilege self.release_collection(user_client, collection_name) # cleanup self.drop_collection(client, collection_name) if with_db: self.using_database(client, "default") self.drop_database(client, db_name) @pytest.mark.parametrize("with_db", [False, True]) def test_milvus_client_verify_grant_collection_insert_privilege(self, host, port, with_db): """ target: test milvus client api verify grant collection insert privilege method: verify grant collection insert privilege expected: verify success """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) collection_name = cf.gen_unique_str(prefix) role_name = cf.gen_unique_str(role_pre) # create user and role self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) # handle database parameter if with_db: db_name = cf.gen_unique_str(prefix) self.create_database(client, db_name) self.using_database(client, db_name) else: db_name = "default" # create collection self.create_collection(client, collection_name, default_dim, consistency_level="Strong") # grant collection insert privilege to role self.grant_privilege(client, role_name, "Collection", "Insert", collection_name, db_name=db_name) # wait for privilege to take effect time.sleep(10) # create new client with the user credentials uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password, db_name=db_name) # test insert privilege rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(user_client, collection_name, rows) # cleanup self.drop_collection(client, collection_name) if with_db: self.using_database(client, "default") self.drop_database(client, db_name) @pytest.mark.parametrize("with_db", [False, True]) def test_milvus_client_verify_grant_collection_delete_privilege(self, host, port, with_db): """ target: test milvus client api verify grant collection delete privilege method: verify grant collection delete privilege expected: verify success """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) collection_name = cf.gen_unique_str(prefix) role_name = cf.gen_unique_str(role_pre) # create user and role self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) # handle database parameter if with_db: db_name = cf.gen_unique_str(prefix) self.create_database(client, db_name) self.using_database(client, db_name) else: db_name = "default" # create collection self.create_collection(client, collection_name, default_dim, consistency_level="Strong") # grant collection delete privilege to role self.grant_privilege(client, role_name, "Collection", "Delete", collection_name, db_name=db_name) # wait for privilege to take effect time.sleep(10) # create new client with the user credentials uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password, db_name=db_name) # test delete privilege delete_expr = f"{default_primary_key_field_name} == 0" self.delete(user_client, collection_name, filter=delete_expr) # cleanup self.drop_collection(client, collection_name) if with_db: self.using_database(client, "default") self.drop_database(client, db_name) def test_milvus_client_new_user_default_owns_public_role_permission(self, host, port): """ target: test milvus client api new user owns public role privilege method: create a role, verify its permission expected: verify success """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) collection_name = cf.gen_unique_str(prefix) collection_name_2 = cf.gen_unique_str(prefix) role_name = cf.gen_unique_str(role_pre) # create user self.create_user(client, user_name=user_name, password=password) # create collection self.create_collection(client, collection_name, default_dim, consistency_level="Strong") # create index index_params, _ = self.prepare_index_params(client) index_params.add_index(field_name=default_vector_field_name) self.create_index(client, collection_name, index_params) # create new client with the user credentials uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) # wait for user to be created time.sleep(10) # test collection permission deny (new user has no privileges) self.load_collection(user_client, collection_name, check_task=CheckTasks.check_permission_deny) self.release_collection(user_client, collection_name, check_task=CheckTasks.check_permission_deny) # test insert permission deny rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(user_client, collection_name, rows, check_task=CheckTasks.check_permission_deny) # test delete permission deny delete_expr = f"{default_primary_key_field_name} == 0" self.delete(user_client, collection_name, filter=delete_expr, check_task=CheckTasks.check_permission_deny) # test search permission deny vectors_to_search = rng.random((1, default_dim)) self.search(user_client, collection_name, vectors_to_search, check_task=CheckTasks.check_permission_deny) # test query permission deny query_expr = f"{default_primary_key_field_name} in [0, 1]" self.query(user_client, collection_name, filter=query_expr, check_task=CheckTasks.check_permission_deny) # test global permission deny self.create_collection( user_client, collection_name_2, default_dim, consistency_level="Strong", check_task=CheckTasks.check_permission_deny, ) self.drop_collection(user_client, collection_name, check_task=CheckTasks.check_permission_deny) self.create_user( user_client, user_name=collection_name, password=password, check_task=CheckTasks.check_permission_deny ) self.create_role(user_client, role_name=role_name, check_task=CheckTasks.check_permission_deny) self.drop_user(user_client, user_name=user_name, check_task=CheckTasks.check_permission_deny) # cleanup self.drop_collection(client, collection_name) self.drop_user(client, user_name=user_name) @pytest.mark.parametrize("role_name", ["admin", "public"]) def test_milvus_client_remove_user_from_default_role(self, role_name, host, port): """ target: test milvus client api remove user from default role (admin or public) method: create a user, add user to admin role or public role, remove user from role expected: remove success """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) # create user self.create_user(client, user_name=user_name, password=password) # grant role to user (add user to role) self.grant_role(client, user_name=user_name, role_name=role_name) # verify user has the role user_info, _ = self.describe_user(client, user_name=user_name) assert user_info["user_name"] == user_name assert role_name in user_info.get("roles", []) # revoke role from user (remove user from role) self.revoke_role(client, user_name=user_name, role_name=role_name) # verify user no longer has the role user_info, _ = self.describe_user(client, user_name=user_name) assert user_info["user_name"] == user_name assert role_name not in user_info.get("roles", []) # cleanup self.drop_user(client, user_name=user_name) @pytest.mark.parametrize("role_name", ["admin", "public"]) def test_milvus_client_drop_admin_and_public_role(self, role_name, host, port): """ target: test milvus client api drop admin and public role fail method: drop admin and public role fail expected: fail to drop """ client = self._client() # verify role exists roles, _ = self.list_roles(client) assert role_name in roles # try to drop default role (should fail) error_msg = f"the role[{role_name}] is a default role, which can't be dropped" self.drop_role( client, role_name=role_name, check_task=CheckTasks.err_res, check_items={ct.err_code: 1401, ct.err_msg: error_msg}, ) # verify role still exists roles, _ = self.list_roles(client) assert role_name in roles def test_milvus_client_add_user_not_exist_role(self, host, port): """ target: test milvus client api add user to not exist role method: create a user, add user to not exist role expected: fail to add """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) # create user self.create_user(client, user_name=user_name, password=password) # verify role does not exist roles, _ = self.list_roles(client) assert role_name not in roles # try to grant non-existent role to user (should fail) error_msg = "not found the role, maybe the role isn't existed or internal system error" self.grant_role( client, user_name=user_name, role_name=role_name, check_task=CheckTasks.err_res, check_items={ct.err_code: 65535, ct.err_msg: error_msg}, ) # cleanup self.drop_user(client, user_name=user_name) def test_milvus_client_remove_user_from_empty_role(self, host, port): """ target: test milvus client api remove not exist user from role method: create new role, remove not exist user from unbind role expected: fail to remove """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) # create user self.create_user(client, user_name=user_name, password=password) # create role self.create_role(client, role_name=role_name) # verify role does not have the user role_info, _ = self.describe_role(client, role_name=role_name) role_users = role_info.get("users", []) assert user_name not in role_users # try to revoke user from role (should fail since user is not in the role) self.revoke_role(client, user_name=user_name, role_name=role_name) # verify user is still not in the role role_info, _ = self.describe_role(client, role_name=role_name) role_users = role_info.get("users", []) assert user_name not in role_users # cleanup self.drop_role(client, role_name=role_name) self.drop_user(client, user_name=user_name) def test_milvus_client_list_grant_by_not_exist_role(self, host, port): """ target: test milvus client api list grants by not exist role method: list grants by not exist role expected: fail to list """ client = self._client() role_name = cf.gen_unique_str(role_pre) # verify role does not exist roles, _ = self.list_roles(client) assert role_name not in roles # try to describe non-existent role (should fail) error_msg = "not found the role, maybe the role isn't existed or internal system error" self.describe_role( client, role_name=role_name, check_task=CheckTasks.err_res, check_items={ct.err_code: 65535, ct.err_msg: error_msg}, ) @pytest.mark.tags(CaseLabel.RBAC) class TestMilvusClientRbacPrivilegeVerify(TestMilvusClientV2Base): """Test case of rbac privilege verification""" def teardown_method(self, method): log.info("[teardown_method] Start teardown RBAC test cases ...") _teardown_rbac(self) super().teardown_method(method) # ==================== P0 Tests ==================== def test_milvus_client_new_user_default_public_role_permission(self, host, port): """ target: test new user with no explicit role has no privileges except list_collections method: create user with no role, verify operations are denied, list_collections succeeds expected: 14+ operations denied, list_collections succeeds """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) collection_name = cf.gen_unique_str(prefix) collection_name_2 = cf.gen_unique_str(prefix) role_name = cf.gen_unique_str(role_pre) # create user (no role assigned) self.create_user(client, user_name=user_name, password=password) # create collection + index for testing self.create_collection(client, collection_name, default_dim, consistency_level="Strong") index_params = self.prepare_index_params(client)[0] index_params.add_index(default_vector_field_name, metric_type="COSINE") self.create_index(client, collection_name, index_params) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) time.sleep(10) # verify operations denied self.load_collection(user_client, collection_name, check_task=CheckTasks.check_permission_deny) self.release_collection(user_client, collection_name, check_task=CheckTasks.check_permission_deny) rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(user_client, collection_name, rows, check_task=CheckTasks.check_permission_deny) delete_expr = f"{default_primary_key_field_name} == 0" self.delete(user_client, collection_name, filter=delete_expr, check_task=CheckTasks.check_permission_deny) vectors_to_search = cf.gen_vectors(1, default_dim) self.search(user_client, collection_name, vectors_to_search, check_task=CheckTasks.check_permission_deny) query_expr = f"{default_primary_key_field_name} in [0, 1]" self.query(user_client, collection_name, filter=query_expr, check_task=CheckTasks.check_permission_deny) self.create_collection(user_client, collection_name_2, default_dim, check_task=CheckTasks.check_permission_deny) self.drop_collection(user_client, collection_name, check_task=CheckTasks.check_permission_deny) self.create_user( user_client, user_name=cf.gen_unique_str(user_pre), password=password, check_task=CheckTasks.check_permission_deny, ) self.create_role(user_client, role_name=role_name, check_task=CheckTasks.check_permission_deny) self.drop_user(user_client, user_name=user_name, check_task=CheckTasks.check_permission_deny) self.drop_role(user_client, role_name="admin", check_task=CheckTasks.check_permission_deny) self.grant_privilege(user_client, "admin", "Global", "All", "*", check_task=CheckTasks.check_permission_deny) self.revoke_privilege(user_client, "admin", "Global", "All", "*", check_task=CheckTasks.check_permission_deny) self.grant_role( user_client, user_name=user_name, role_name="admin", check_task=CheckTasks.check_permission_deny ) # list_collections should succeed (public role) res, _ = self.list_collections(user_client) assert isinstance(res, list) # cleanup self.drop_collection(client, collection_name) def test_milvus_client_verify_grant_collection_insert_privilege(self, host, port): """ target: test grant Insert on specific collection only method: create 2 collections, grant Insert on col_a only expected: user insert col_a succeeds, col_b denied, create_index on col_a denied """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) col_a = cf.gen_unique_str(prefix) col_b = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.create_collection(client, col_a, default_dim, consistency_level="Strong") self.create_collection(client, col_b, default_dim, consistency_level="Strong") self.grant_privilege(client, role_name, "Collection", "Insert", col_a) time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] # insert into col_a succeeds self.insert(user_client, col_a, rows) # insert into col_b denied self.insert(user_client, col_b, rows, check_task=CheckTasks.check_permission_deny) # create_index on col_a denied (no CreateIndex privilege) index_params = self.prepare_index_params(user_client)[0] index_params.add_index(default_vector_field_name, metric_type="COSINE") self.create_index(user_client, col_a, index_params, check_task=CheckTasks.check_permission_deny) # cleanup self.drop_collection(client, col_a) self.drop_collection(client, col_b) def test_milvus_client_verify_grant_collection_delete_privilege(self, host, port): """ target: test grant Delete privilege on collection method: root creates collection + inserts, grant Delete, user deletes expected: user delete succeeds """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.create_collection(client, collection_name, default_dim, consistency_level="Strong") rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(client, collection_name, rows) self.grant_privilege(client, role_name, "Collection", "Delete", collection_name) time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) delete_expr = f"{default_primary_key_field_name} in [0, 1, 2]" self.delete(user_client, collection_name, filter=delete_expr) # cleanup self.drop_collection(client, collection_name) def test_milvus_client_verify_grant_collection_search_privilege(self, host, port): """ target: test grant Search privilege on collection method: root creates + inserts + loads, grant Search, user searches expected: user search returns results """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.create_collection(client, collection_name, default_dim, consistency_level="Strong") rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(client, collection_name, rows) index_params = self.prepare_index_params(client)[0] index_params.add_index(default_vector_field_name, metric_type="COSINE") self.create_index(client, collection_name, index_params) self.load_collection(client, collection_name) self.grant_privilege(client, role_name, "Collection", "Search", collection_name) time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) vectors_to_search = cf.gen_vectors(1, default_dim) res, _ = self.search(user_client, collection_name, vectors_to_search) assert len(res) > 0 # cleanup self.drop_collection(client, collection_name) def test_milvus_client_verify_grant_collection_query_privilege(self, host, port): """ target: test grant Query privilege on collection method: root creates + inserts + loads, grant Query, user queries expected: user query returns results """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.create_collection(client, collection_name, default_dim, consistency_level="Strong") rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(client, collection_name, rows) index_params = self.prepare_index_params(client)[0] index_params.add_index(default_vector_field_name, metric_type="COSINE") self.create_index(client, collection_name, index_params) self.load_collection(client, collection_name) self.grant_privilege(client, role_name, "Collection", "Query", collection_name) time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) query_expr = f"{default_primary_key_field_name} in [0, 1, 2]" res, _ = self.query(user_client, collection_name, filter=query_expr) assert len(res) > 0 # cleanup self.drop_collection(client, collection_name) def test_milvus_client_verify_grant_collection_load_privilege(self, host, port): """ target: test grant Load + GetLoadingProgress privilege on collection method: root creates + inserts + creates index, grant Load + GetLoadingProgress, user loads expected: user load succeeds """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.create_collection(client, collection_name, default_dim, consistency_level="Strong") rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(client, collection_name, rows) index_params = self.prepare_index_params(client)[0] index_params.add_index(default_vector_field_name, metric_type="COSINE") self.create_index(client, collection_name, index_params) self.grant_privilege(client, role_name, "Collection", "Load", collection_name) self.grant_privilege(client, role_name, "Collection", "GetLoadingProgress", collection_name) time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) self.load_collection(user_client, collection_name) # cleanup self.drop_collection(client, collection_name) def test_milvus_client_verify_grant_collection_release_privilege(self, host, port): """ target: test grant Release privilege on collection method: root creates + loads, grant Release, user releases expected: user release succeeds """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.create_collection(client, collection_name, default_dim, consistency_level="Strong") self.load_collection(client, collection_name) self.grant_privilege(client, role_name, "Collection", "Release", collection_name) time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) self.release_collection(user_client, collection_name) # cleanup self.drop_collection(client, collection_name) def test_milvus_client_verify_global_all_privilege(self, host, port): """ target: test grant Global All privilege method: grant Global All *, user can create/insert/drop collection + create/drop user + create/drop role expected: all operations succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.grant_privilege(client, role_name, "Global", "All", "*") time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) # create and drop collection coll_name = cf.gen_unique_str(prefix) self.create_collection(user_client, coll_name, default_dim, consistency_level="Strong") rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(user_client, coll_name, rows) self.drop_collection(user_client, coll_name) # create and drop user tmp_user = cf.gen_unique_str(user_pre) tmp_password = cf.gen_str_by_length(contain_numbers=True) self.create_user(user_client, user_name=tmp_user, password=tmp_password) self.drop_user(user_client, user_name=tmp_user) # create and drop role tmp_role = cf.gen_unique_str(role_pre) self.create_role(user_client, role_name=tmp_role) self.drop_role(user_client, role_name=tmp_role) def test_milvus_client_role_revoke_user_privilege(self, host, port): """ target: test grant and revoke Global UpdateUser privilege method: grant UpdateUser, user updates OTHER user's password, revoke, denied expected: update other's password succeeds then denied after revoke """ client = self._client() user_name = cf.gen_unique_str(user_pre) target_user = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) target_password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) self.create_user(client, user_name=user_name, password=password) self.create_user(client, user_name=target_user, password=target_password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.grant_privilege_v2(client, role_name, "UpdateUser", collection_name="*", db_name="*") time.sleep(20) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) # user can update OTHER user's password (requires UpdateUser privilege) new_target_password = cf.gen_str_by_length(contain_numbers=True) self.update_password( user_client, user_name=target_user, old_password=target_password, new_password=new_target_password ) # revoke UpdateUser self.revoke_privilege_v2(client, role_name, "UpdateUser", collection_name="*", db_name="*") time.sleep(20) # reconnect to pick up revoked privilege user_client2, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) another_password = cf.gen_str_by_length(contain_numbers=True) self.update_password( user_client2, user_name=target_user, old_password=new_target_password, new_password=another_password, check_task=CheckTasks.check_permission_deny, ) def test_milvus_client_verify_grant_wildcard_object_name(self, host, port): """ target: test grant Insert with wildcard object name method: grant Collection Insert on "*", root creates 2 collections, user inserts both expected: user inserts both succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.grant_privilege(client, role_name, "Collection", "Insert", "*") time.sleep(10) col_a = cf.gen_unique_str(prefix) col_b = cf.gen_unique_str(prefix) self.create_collection(client, col_a, default_dim, consistency_level="Strong") self.create_collection(client, col_b, default_dim, consistency_level="Strong") uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(user_client, col_a, rows) self.insert(user_client, col_b, rows) # cleanup self.drop_collection(client, col_a) self.drop_collection(client, col_b) def test_milvus_client_verify_grant_wildcard_privilege(self, host, port): """ target: test grant Collection * (all privileges) on specific collection method: grant Collection * on col, root inserts + loads, user can insert/search/query expected: all operations succeed on the collection """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.create_collection(client, collection_name, default_dim, consistency_level="Strong") rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(client, collection_name, rows) index_params = self.prepare_index_params(client)[0] index_params.add_index(default_vector_field_name, metric_type="COSINE") self.create_index(client, collection_name, index_params) self.load_collection(client, collection_name) self.grant_privilege_v2(client, role_name, "CollectionReadWrite", collection_name=collection_name) time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) # user can insert new_rows = [ { default_primary_key_field_name: i + default_nb, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: (i + default_nb) * 1.0, default_string_field_name: str(i + default_nb), } for i in range(10) ] self.insert(user_client, collection_name, new_rows) # user can search vectors_to_search = cf.gen_vectors(1, default_dim) res, _ = self.search(user_client, collection_name, vectors_to_search) assert len(res) > 0 # user can query query_expr = f"{default_primary_key_field_name} in [0, 1, 2]" res, _ = self.query(user_client, collection_name, filter=query_expr) assert len(res) > 0 # cleanup self.drop_collection(client, collection_name) # ==================== P1 Tests ==================== def test_milvus_client_verify_grant_create_index_privilege(self, host, port): """ target: test grant CreateIndex privilege on collection method: root creates collection, grant CreateIndex, user creates index expected: user create_index succeeds """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.create_collection(client, collection_name, default_dim, consistency_level="Strong") self.grant_privilege(client, role_name, "Collection", "CreateIndex", collection_name) time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) index_params = self.prepare_index_params(user_client)[0] index_params.add_index(default_vector_field_name, metric_type="COSINE") self.create_index(user_client, collection_name, index_params) # cleanup self.drop_collection(client, collection_name) def test_milvus_client_verify_grant_drop_index_privilege(self, host, port): """ target: test grant DropIndex privilege on collection method: root creates collection + index, grant DropIndex, user drops index expected: user drop_index succeeds """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) # Use schema mode to avoid auto-load, then manually create index schema = self.create_schema(client)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True) schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) self.create_collection(client, collection_name, schema=schema) index_params = self.prepare_index_params(client)[0] index_params.add_index(default_vector_field_name, metric_type="COSINE") self.create_index(client, collection_name, index_params) # collection is NOT loaded, so drop_index is allowed # grant DropIndex + DescribeCollection via v2 API self.grant_privilege_v2(client, role_name, "DropIndex", collection_name=collection_name) self.grant_privilege_v2(client, role_name, "DescribeCollection", collection_name=collection_name) time.sleep(20) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) self.drop_index(user_client, collection_name, default_vector_field_name) # cleanup self.drop_collection(client, collection_name) def test_milvus_client_verify_grant_compaction_privilege(self, host, port): """ target: test grant Compaction privilege on collection method: root creates collection + inserts, grant Compaction, user compacts expected: user compact succeeds """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.create_collection(client, collection_name, default_dim, consistency_level="Strong") rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(client, collection_name, rows) self.grant_privilege(client, role_name, "Collection", "Compaction", collection_name) time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) self.compact(user_client, collection_name) # cleanup self.drop_collection(client, collection_name) def test_milvus_client_verify_grant_flush_privilege(self, host, port): """ target: test grant Flush privilege on collection method: root creates collection + inserts, grant Flush, user flushes expected: user flush succeeds """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.create_collection(client, collection_name, default_dim, consistency_level="Strong") rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(client, collection_name, rows) self.grant_privilege(client, role_name, "Collection", "Flush", collection_name) time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) self.flush(user_client, collection_name) # cleanup self.drop_collection(client, collection_name) def test_milvus_client_verify_grant_global_create_collection_privilege(self, host, port): """ target: test grant Global CreateCollection privilege method: grant CreateCollection, user creates collection expected: user create_collection succeeds """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) # FIX: CreateCollection is a cluster-level privilege, needs db_name="*" self.grant_privilege_v2(client, role_name, "CreateCollection", collection_name="*", db_name="*") time.sleep(15) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) coll_name = cf.gen_unique_str(prefix) # Use schema mode to avoid auto-index/load which needs extra privileges schema = self.create_schema(user_client)[0] schema.add_field(default_primary_key_field_name, DataType.INT64, is_primary=True) schema.add_field(default_vector_field_name, DataType.FLOAT_VECTOR, dim=default_dim) self.create_collection(user_client, coll_name, schema=schema) # cleanup self.drop_collection(client, coll_name) def test_milvus_client_verify_grant_global_drop_collection_privilege(self, host, port): """ target: test grant Global DropCollection privilege method: root creates collection, grant DropCollection, user drops expected: user drop_collection succeeds """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.create_collection(client, collection_name, default_dim, consistency_level="Strong") self.grant_privilege(client, role_name, "Global", "DropCollection", "*") time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) self.drop_collection(user_client, collection_name) def test_milvus_client_verify_grant_global_create_ownership_privilege(self, host, port): """ target: test grant Global CreateOwnership privilege method: grant CreateOwnership, user can create user and create role expected: user create_user and create_role succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.grant_privilege(client, role_name, "Global", "CreateOwnership", "*") time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) tmp_user = cf.gen_unique_str(user_pre) tmp_password = cf.gen_str_by_length(contain_numbers=True) self.create_user(user_client, user_name=tmp_user, password=tmp_password) tmp_role = cf.gen_unique_str(role_pre) self.create_role(user_client, role_name=tmp_role) # cleanup self.drop_role(client, role_name=tmp_role) self.drop_user(client, user_name=tmp_user) def test_milvus_client_verify_grant_global_drop_ownership_privilege(self, host, port): """ target: test grant Global DropOwnership privilege method: root creates user + role, grant DropOwnership, user drops them expected: user drop_user and drop_role succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.grant_privilege(client, role_name, "Global", "DropOwnership", "*") time.sleep(10) # root creates targets to drop tmp_user = cf.gen_unique_str(user_pre) tmp_password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=tmp_user, password=tmp_password) tmp_role = cf.gen_unique_str(role_pre) self.create_role(client, role_name=tmp_role) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) self.drop_user(user_client, user_name=tmp_user) self.drop_role(user_client, role_name=tmp_role) def test_milvus_client_verify_grant_global_select_ownership_privilege(self, host, port): """ target: test grant Global SelectOwnership privilege method: grant SelectOwnership, user can list_users and list_roles expected: user list_users and list_roles succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.grant_privilege(client, role_name, "Global", "SelectOwnership", "*") time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) users, _ = self.list_users(user_client) assert len(users) >= 1 roles, _ = self.list_roles(user_client) assert len(roles) >= 2 def test_milvus_client_verify_grant_global_manage_ownership_privilege(self, host, port): """ target: test grant Global ManageOwnership privilege method: grant ManageOwnership, user can grant_role and revoke_role expected: user grant_role and revoke_role succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.grant_privilege(client, role_name, "Global", "ManageOwnership", "*") time.sleep(10) # root creates a second user and role tmp_user = cf.gen_unique_str(user_pre) tmp_password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=tmp_user, password=tmp_password) tmp_role = cf.gen_unique_str(role_pre) self.create_role(client, role_name=tmp_role) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) # user can grant role self.grant_role(user_client, user_name=tmp_user, role_name=tmp_role) # user can revoke role self.revoke_role(user_client, user_name=tmp_user, role_name=tmp_role) # cleanup self.drop_role(client, role_name=tmp_role) self.drop_user(client, user_name=tmp_user) def test_milvus_client_verify_grant_user_update_privilege(self, host, port): """ target: test grant Global UpdateUser privilege method: grant UpdateUser, user can update own password expected: user update_password succeeds """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) self.grant_privilege_v2(client, role_name, "UpdateUser", collection_name="*") time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) new_password = cf.gen_str_by_length(contain_numbers=True) self.update_password(user_client, user_name=user_name, old_password=password, new_password=new_password) def test_milvus_client_verify_grant_user_select_privilege(self, host, port): """ target: test grant Global SelectUser privilege method: grant SelectUser, user can describe_user and list_users expected: user describe_user and list_users succeed """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) # SelectUser allows describe_user; SelectOwnership allows list_users self.grant_privilege_v2(client, role_name, "SelectUser", collection_name="*", db_name="*") self.grant_privilege_v2(client, role_name, "SelectOwnership", collection_name="*", db_name="*") time.sleep(20) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) res, _ = self.describe_user(user_client, user_name=user_name) assert res["user_name"] == user_name users, _ = self.list_users(user_client) assert user_name in users @pytest.mark.skip("public role privileges may be granted via built-in groups and cannot be individually revoked") def test_milvus_client_revoke_public_role_privilege(self, host, port): """ target: test revoke privilege from public role method: revoke ShowCollections from public, user denied list_collections, re-grant expected: user denied after revoke, succeeds after re-grant """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=user_name, password=password) try: # revoke ShowCollections from public role self.revoke_privilege_v2(client, "public", "ShowCollections", collection_name="*", db_name="*") time.sleep(20) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) self.list_collections(user_client, check_task=CheckTasks.check_permission_deny) finally: # re-grant to restore public role self.grant_privilege_v2(client, "public", "ShowCollections", collection_name="*", db_name="*") time.sleep(20) def test_milvus_client_revoke_user_after_delete_user(self, host, port): """ target: test recreate user after delete method: create user, bind role, delete user, recreate with same name, login with new password expected: new user can login with new password """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) # delete user self.drop_user(client, user_name=user_name) time.sleep(10) # recreate user with same name but different password new_password = cf.gen_str_by_length(contain_numbers=True) self.create_user(client, user_name=user_name, password=new_password) time.sleep(20) # login with new password should succeed uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=new_password) res, _ = self.list_collections(user_client) assert isinstance(res, list) def test_milvus_client_grant_connect_privilege(self, host, port): """ target: test grant SelectUser privilege allows list_users and list_collections method: grant SelectUser, user can list_users and list_collections but not create_collection expected: list_users ok, list_collections ok, create_collection denied """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) # SelectUser + SelectOwnership for list_users self.grant_privilege_v2(client, role_name, "SelectUser", collection_name="*", db_name="*") self.grant_privilege_v2(client, role_name, "SelectOwnership", collection_name="*", db_name="*") time.sleep(20) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) users, _ = self.list_users(user_client) assert len(users) >= 1 res, _ = self.list_collections(user_client) assert isinstance(res, list) coll_name = cf.gen_unique_str(prefix) self.create_collection(user_client, coll_name, default_dim, check_task=CheckTasks.check_permission_deny) def test_milvus_client_public_role_privilege_all_dbs(self, host, port): """ target: test public role user can list_collections in default db only method: create db_a and db_b, public user lists in default db and cannot switch to other dbs expected: list_collections succeeds in default db and using_database is denied in db_a/db_b """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) db_a = cf.gen_unique_str(prefix) db_b = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_database(client, db_a) self.create_database(client, db_b) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) # list_collections in default db res, _ = self.list_collections(user_client) assert isinstance(res, list) # public role cannot switch to another db without DescribeDatabase privilege self.using_database( user_client, db_a, check_task=CheckTasks.err_res, check_items={ct.err_code: 7, ct.err_msg: "PrivilegeDescribeDatabase: permission deny"}, ) self.using_database( user_client, db_b, check_task=CheckTasks.err_res, check_items={ct.err_code: 7, ct.err_msg: "PrivilegeDescribeDatabase: permission deny"}, ) # cleanup self.using_database(client, "default") self.drop_database(client, db_a) self.drop_database(client, db_b) def test_milvus_client_built_in_privilege_groups_e2e(self, host, port): """ target: test built-in privilege groups end to end method: grant CollectionReadOnly, search ok, insert denied, switch to ReadWrite, insert ok, then custom group with Insert, grant, verify, remove Insert, verify denied expected: privilege groups work correctly """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) # create collection, insert, index, load self.create_collection(client, collection_name, default_dim, consistency_level="Strong") rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(client, collection_name, rows) index_params = self.prepare_index_params(client)[0] index_params.add_index(default_vector_field_name, metric_type="COSINE") self.create_index(client, collection_name, index_params) self.load_collection(client, collection_name) # grant CollectionReadOnly self.grant_privilege_v2(client, role_name, "CollectionReadOnly", collection_name=collection_name) # FIX: increase sleep times throughout for privilege propagation time.sleep(20) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) # search ok vectors_to_search = cf.gen_vectors(1, default_dim) res, _ = self.search(user_client, collection_name, vectors_to_search) assert len(res) > 0 # insert denied (ReadOnly) new_rows = [ { default_primary_key_field_name: i + default_nb, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: (i + default_nb) * 1.0, default_string_field_name: str(i + default_nb), } for i in range(10) ] self.insert(user_client, collection_name, new_rows, check_task=CheckTasks.check_permission_deny) # revoke CollectionReadOnly, grant CollectionReadWrite self.revoke_privilege_v2(client, role_name, "CollectionReadOnly", collection_name=collection_name) self.grant_privilege_v2(client, role_name, "CollectionReadWrite", collection_name=collection_name) time.sleep(20) # insert ok (ReadWrite) user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) self.insert(user_client, collection_name, new_rows) # revoke CollectionReadWrite self.revoke_privilege_v2(client, role_name, "CollectionReadWrite", collection_name=collection_name) time.sleep(20) # create custom privilege group with Insert pg_name = cf.gen_unique_str(prefix) self.create_privilege_group(client, privilege_group=pg_name) self.add_privileges_to_group(client, privilege_group=pg_name, privileges=["Insert"]) self.grant_privilege_v2(client, role_name, pg_name, collection_name=collection_name) time.sleep(20) user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) more_rows = [ { default_primary_key_field_name: i + default_nb + 10, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: (i + default_nb + 10) * 1.0, default_string_field_name: str(i + default_nb + 10), } for i in range(10) ] self.insert(user_client, collection_name, more_rows) # revoke the privilege group grant first, then remove privilege from group self.revoke_privilege_v2(client, role_name, pg_name, collection_name=collection_name) self.remove_privileges_from_group(client, privilege_group=pg_name, privileges=["Insert"]) time.sleep(30) # insert denied (use different PKs to avoid duplicate key issues) user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) denied_rows = [ { default_primary_key_field_name: i + default_nb + 20, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: (i + default_nb + 20) * 1.0, default_string_field_name: str(i + default_nb + 20), } for i in range(10) ] self.insert(user_client, collection_name, denied_rows, check_task=CheckTasks.check_permission_deny) # cleanup self.drop_privilege_group(client, privilege_group=pg_name) self.drop_collection(client, collection_name) @pytest.mark.tags(CaseLabel.RBAC) class TestMilvusClientRbacPrivilegeGroup(TestMilvusClientV2Base): """Test case of rbac privilege group interface""" def teardown_method(self, method): log.info("[teardown_method] Start teardown privilege group test cases ...") _teardown_rbac(self) super().teardown_method(method) def test_milvus_client_create_large_numbers_privilege_groups(self, host, port): """ target: test create large number of privilege groups method: create 100 privilege groups, verify all exist, drop all expected: all operations succeed """ client = self._client() pg_names = [] for i in range(100): pg_name = cf.gen_unique_str(prefix) self.create_privilege_group(client, privilege_group=pg_name) pg_names.append(pg_name) # verify all groups exist pgs, _ = self.list_privilege_groups(client) pg_list = [pg.get("privilege_group", pg) if isinstance(pg, dict) else str(pg) for pg in pgs] for pg_name in pg_names: assert pg_name in pg_list # drop all groups for pg_name in pg_names: self.drop_privilege_group(client, privilege_group=pg_name) # verify all groups are dropped pgs, _ = self.list_privilege_groups(client) pg_list = [pg.get("privilege_group", pg) if isinstance(pg, dict) else str(pg) for pg in pgs] for pg_name in pg_names: assert pg_name not in pg_list @pytest.mark.tags(CaseLabel.RBAC) class TestMilvusClientRbacPrefixIsolation(TestMilvusClientV2Base): """PR #48053 / Issue #47998: RBAC prefix isolation tests. Bug: When two usernames have a prefix relationship (e.g. user1 / user11), getRolesByUsername returns empty-string roles due to path.Join stripping trailing slash in etcd LoadWithPrefix. """ DIM = 128 def teardown_method(self, method): """Clean up entities created by this test, then call super() for connection cleanup.""" if not hasattr(self, "_p"): super().teardown_method(method) return log.info(f"[pr48053 teardown] cleaning up {method.__name__}, prefix={self._p}") client = self._client() # Use force_drop to skip manual revoke of privileges. # Order: drop roles (force) → drop users → drop privilege groups # → drop aliases → drop collections → drop databases for role in [self._role1, self._role10, self._role1_read]: try: client.drop_role(role, force_drop=True) except Exception: pass for user in [self._user1, self._user11, self._user1_ro]: try: client.drop_user(user) except Exception: pass for pg in [self._pg1, self._pg10, self._pg1_ext]: try: client.drop_privilege_group(pg) except Exception: pass for alias in [self._alias1, self._alias10, self._alias1_bak]: try: client.drop_alias(alias) except Exception: pass for col in [self._col, self._col_v2, self._collection]: try: client.drop_collection(col) except Exception: pass for db in [self._db1, self._db10]: try: client.using_database(db) client.drop_collection("inner_col") except Exception: pass try: client.using_database("default") client.drop_database(db) except Exception: pass super().teardown_method(method) def test_prefix_isolation_all(self, host, port): """All UpgradeCheck cases (#01~#26) in a single test method. Before upgrade (bug present): - #03 DescribeUser(user1): FAIL (empty-string role) - #23 ListDatabases as user1: FAIL (role entity invalid) After upgrade (bug fixed): all pass. """ client = self._client() uri = f"http://{host}:{port}" # ============================================================ # Setup: create all test data with random prefix # ============================================================ p = cf.gen_unique_str("pr48053") self._p = p # Names with prefix collision self._user1 = f"{p}_user1" self._user11 = f"{p}_user11" self._user1_ro = f"{p}_user1_ro" self._pw1 = cf.gen_str_by_length(contain_numbers=True) self._pw11 = cf.gen_str_by_length(contain_numbers=True) self._pw1_ro = cf.gen_str_by_length(contain_numbers=True) self._role1 = f"{p}_role1" self._role10 = f"{p}_role10" self._role1_read = f"{p}_role1_read" self._col = f"{p}_col" self._col_v2 = f"{p}_col_v2" self._collection = f"{p}_collection" self._alias1 = f"{p}_alias1" self._alias10 = f"{p}_alias10" self._alias1_bak = f"{p}_alias1_bak" self._db1 = f"{p}_db1" self._db10 = f"{p}_db10" self._pg1 = f"{p}_pg1" self._pg10 = f"{p}_pg10" self._pg1_ext = f"{p}_pg1_extended" dim = self.DIM # Collections for col_name in [self._col, self._col_v2, self._collection]: schema, _ = self.create_schema(client) schema.add_field("id", DataType.INT64, is_primary=True) schema.add_field("vec", DataType.FLOAT_VECTOR, dim=dim) self.create_collection(client, col_name, schema=schema, force_teardown=False) self.insert(client, col_name, [{"id": i, "vec": np.random.random(dim).tolist()} for i in range(10)]) idx = client.prepare_index_params() idx.add_index(field_name="vec", index_type="FLAT", metric_type="COSINE") self.create_index(client, col_name, idx) self.load_collection(client, col_name) # Aliases self.create_alias(client, self._col, self._alias1) self.create_alias(client, self._col_v2, self._alias10) self.create_alias(client, self._collection, self._alias1_bak) # Databases + inner collections for db_name in [self._db1, self._db10]: self.create_database(client, db_name) self.using_database(client, db_name) schema, _ = self.create_schema(client) schema.add_field("id", DataType.INT64, is_primary=True) schema.add_field("vec", DataType.FLOAT_VECTOR, dim=dim) self.create_collection(client, "inner_col", schema=schema, force_teardown=False) self.insert(client, "inner_col", [{"id": i, "vec": np.random.random(dim).tolist()} for i in range(10)]) idx = client.prepare_index_params() idx.add_index(field_name="vec", index_type="FLAT", metric_type="COSINE") self.create_index(client, "inner_col", idx) self.load_collection(client, "inner_col") self.using_database(client, "default") # Roles for role in [self._role1, self._role10, self._role1_read]: self.create_role(client, role) # Users + bindings self.create_user(client, self._user1, self._pw1) self.create_user(client, self._user11, self._pw11) self.create_user(client, self._user1_ro, self._pw1_ro) self.grant_role(client, self._user1, self._role1) self.grant_role(client, self._user11, self._role10) self.grant_role(client, self._user1_ro, self._role1_read) # Privilege groups self.create_privilege_group(client, self._pg1) self.add_privileges_to_group(client, self._pg1, ["Search", "Query"]) self.create_privilege_group(client, self._pg10) self.add_privileges_to_group(client, self._pg10, ["Insert", "Delete", "Upsert"]) self.create_privilege_group(client, self._pg1_ext) self.add_privileges_to_group(client, self._pg1_ext, ["Search", "Query", "Load", "Release"]) # Grants (CreateCollection db must be "default" not "*" for #23 to reproduce) grants = [ (self._role1, "Search", self._col, "default"), (self._role1, "Query", self._col, "default"), (self._role10, "Insert", self._col_v2, "default"), (self._role10, "Search", self._col_v2, "default"), (self._role1_read, "Search", self._collection, "default"), (self._role1, "CreateCollection", "*", "default"), (self._role1, "Search", "inner_col", self._db1), (self._role10, "Search", "inner_col", self._db10), (self._role1_read, self._pg1, self._col_v2, "default"), ] for role, priv, col, db in grants: self.grant_privilege_v2(client, role, priv, col, db) time.sleep(2) log.info(f"[pr48053] setup complete, prefix={p}") # ============================================================ # #01~#10: RBAC metadata integrity # ============================================================ # #01: ListUsers users, _ = self.list_users(client) for u in [self._user1, self._user11, self._user1_ro]: assert u in users, f"#01: {u} not in {users}" # #02: ListRoles roles, _ = self.list_roles(client) for r in [self._role1, self._role10, self._role1_read]: assert r in roles, f"#02: {r} not in {roles}" # #03: DescribeUser(user1) - BUG POINT # Bug #47998: user1 is prefix of user11/user1_ro, returns ['role1', '', ''] result, _ = self.describe_user(client, self._user1) roles_u1 = list(result.get("roles", [])) assert "" not in roles_u1, f"#03 Bug #47998: empty-string role, roles={roles_u1}" assert self._role1 in roles_u1, f"#03: should contain {self._role1}, roles={roles_u1}" # #04: DescribeUser(user11) result, _ = self.describe_user(client, self._user11) roles_u11 = list(result.get("roles", [])) assert self._role10 in roles_u11 and self._role1 not in roles_u11 and "" not in roles_u11, ( f"#04: roles={roles_u11}" ) # #05: DescribeUser(user1_ro) result, _ = self.describe_user(client, self._user1_ro) roles_uro = list(result.get("roles", [])) assert self._role1_read in roles_uro and self._role1 not in roles_uro and "" not in roles_uro, ( f"#05: roles={roles_uro}" ) # #06: DescribeRole(role1) no col_v2 result, _ = self.describe_role(client, self._role1) privs = result.get("privileges", []) col_v2_privs = [p for p in privs if self._col_v2 in str(p.get("object_name", ""))] assert len(col_v2_privs) == 0, f"#06: role1 should not have col_v2: {col_v2_privs}" assert len(privs) > 0, "#06: role1 should have privileges" # #07: DescribeRole(role10) no col result, _ = self.describe_role(client, self._role10) privs = result.get("privileges", []) col_only = [p for p in privs if p.get("object_name") == self._col] assert len(col_only) == 0, f"#07: role10 should not have col: {col_only}" # #08: DescribeRole(role1_read) no role1's Query/CreateCollection result, _ = self.describe_role(client, self._role1_read) privs = result.get("privileges", []) has_query = any(p.get("privilege") == "Query" and p.get("object_name") == self._col for p in privs) has_create = any(p.get("privilege") == "CreateCollection" for p in privs) assert not has_query and not has_create, f"#08: Query={has_query}, CreateCollection={has_create}" # #09: ListPrivilegeGroups groups, _ = self.list_privilege_groups(client) names = [g.get("privilege_group", g.get("group_name", "")) for g in groups] for pg in [self._pg1, self._pg10, self._pg1_ext]: assert pg in names, f"#09: {pg} not in {names}" # #10: DescribeRole(role1_read) has pg1 on col_v2 result, _ = self.describe_role(client, self._role1_read) privs = result.get("privileges", []) has_pg1 = any( self._pg1 in str(p.get("privilege", "")) and self._col_v2 in str(p.get("object_name", "")) for p in privs ) assert has_pg1, f"#10: should have {self._pg1} on {self._col_v2}, privs={privs}" # ============================================================ # #11~#20: Permission isolation # Uses init_milvus_client per user + check_task for permission checks # ============================================================ # #11: user1 Search col -> success uc1, _ = self.init_milvus_client(uri=uri, user=self._user1, password=self._pw1) self.search(uc1, self._col, data=[np.random.random(dim).tolist()]) # #12: user1 Insert col_v2 -> denied self.insert( uc1, self._col_v2, [{"id": 9999, "vec": np.random.random(dim).tolist()}], check_task=CheckTasks.check_permission_deny, ) # #13: user1 Search col_v2 -> denied self.search( uc1, self._col_v2, data=[np.random.random(dim).tolist()], check_task=CheckTasks.check_permission_deny ) # #14: user11 Search col -> denied uc11, _ = self.init_milvus_client(uri=uri, user=self._user11, password=self._pw11) self.search(uc11, self._col, data=[np.random.random(dim).tolist()], check_task=CheckTasks.check_permission_deny) # #15: user11 Query col -> denied self.query(uc11, self._col, filter="id >= 0", check_task=CheckTasks.check_permission_deny) # #16: user1_ro Insert col -> denied uc_ro, _ = self.init_milvus_client(uri=uri, user=self._user1_ro, password=self._pw1_ro) self.insert( uc_ro, self._col, [{"id": 9999, "vec": np.random.random(dim).tolist()}], check_task=CheckTasks.check_permission_deny, ) # #17: user1_ro Search col_v2 -> success via pg1 self.search(uc_ro, self._col_v2, data=[np.random.random(dim).tolist()]) # #18: user1_ro Insert col_v2 -> denied (pg1 has no Insert) self.insert( uc_ro, self._col_v2, [{"id": 9999, "vec": np.random.random(dim).tolist()}], check_task=CheckTasks.check_permission_deny, ) # #19: DescribeRole(role1) grants no col_v2 result, _ = self.describe_role(client, self._role1) col_v2 = [p for p in result.get("privileges", []) if self._col_v2 in str(p)] assert len(col_v2) == 0, f"#19: role1 should not have col_v2: {col_v2}" # #20: DescribeRole(role10) grants no col result, _ = self.describe_role(client, self._role10) col_only = [p for p in result.get("privileges", []) if p.get("object_name") == self._col] assert len(col_only) == 0, f"#20: role10 should not have col: {col_only}" # ============================================================ # #21~#22: Alias # ============================================================ # #21: ListAliases all_aliases = [] for col_name in [self._col, self._col_v2, self._collection]: result, _ = self.list_aliases(client, col_name) if isinstance(result, dict): all_aliases.extend(result.get("aliases", [])) elif isinstance(result, list): all_aliases.extend(result) for alias in [self._alias1, self._alias10, self._alias1_bak]: assert alias in all_aliases, f"#21: {alias} not in {all_aliases}" # #22: Search via alias1 result, _ = self.search(client, self._alias1, data=[np.random.random(dim).tolist()]) assert len(result) > 0, "#22: search via alias1 returned no results" # ============================================================ # #23~#26: Multi-DB # ============================================================ # #23: user1 ListDatabases - BUG POINT # Bug #47998: getCurrentUserVisibleDatabases -> getRolesByUsername # -> empty role -> SelectGrant error user_client, _ = self.init_milvus_client(uri=uri, user=self._user1, password=self._pw1) self.list_databases(user_client) # #25: user1 Search db1.inner_col -> success uc1_db1, _ = self.init_milvus_client(uri=uri, user=self._user1, password=self._pw1, db_name=self._db1) self.search(uc1_db1, "inner_col", data=[np.random.random(dim).tolist()]) # #26: user1 Search db10.inner_col -> denied uc1_db10, _ = self.init_milvus_client(uri=uri, user=self._user1, password=self._pw1, db_name=self._db10) self.search( uc1_db10, "inner_col", data=[np.random.random(dim).tolist()], check_task=CheckTasks.check_permission_deny ) log.info(f"[pr48053] all checks passed, prefix={p}") @pytest.mark.tags(CaseLabel.RBAC) class TestMilvusClientRbacGrantV2(TestMilvusClientV2Base): """Test case of rbac grant v2 interface""" def teardown_method(self, method): log.info("[teardown_method] Start teardown grant v2 test cases ...") _teardown_rbac(self) super().teardown_method(method) def test_milvus_client_grant_revoke_v2_duplicate_privilege_and_privilege_group(self, host, port): """ target: test grant Search + CollectionReadOnly, revoke Search, still ok via group, then revoke group method: grant Search + CollectionReadOnly, search ok, revoke Search, still ok, revoke group, denied expected: search ok while group covers, denied after revoking both """ client = self._client() user_name = cf.gen_unique_str(user_pre) password = cf.gen_str_by_length(contain_numbers=True) role_name = cf.gen_unique_str(role_pre) collection_name = cf.gen_unique_str(prefix) self.create_user(client, user_name=user_name, password=password) self.create_role(client, role_name=role_name) self.grant_role(client, user_name=user_name, role_name=role_name) # create collection, insert, index, load self.create_collection(client, collection_name, default_dim, consistency_level="Strong") rng = np.random.default_rng(seed=19530) rows = [ { default_primary_key_field_name: i, default_vector_field_name: list(rng.random((1, default_dim))[0]), default_float_field_name: i * 1.0, default_string_field_name: str(i), } for i in range(default_nb) ] self.insert(client, collection_name, rows) index_params = self.prepare_index_params(client)[0] index_params.add_index(default_vector_field_name, metric_type="COSINE") self.create_index(client, collection_name, index_params) self.load_collection(client, collection_name) # grant Search (individual) + CollectionReadOnly (group) self.grant_privilege_v2(client, role_name, "Search", collection_name=collection_name) self.grant_privilege_v2(client, role_name, "CollectionReadOnly", collection_name=collection_name) time.sleep(10) uri = f"http://{host}:{port}" user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) vectors_to_search = cf.gen_vectors(1, default_dim) # search ok res, _ = self.search(user_client, collection_name, vectors_to_search) assert len(res) > 0 # revoke Search (individual) — group still covers self.revoke_privilege_v2(client, role_name, "Search", collection_name=collection_name) time.sleep(10) user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) res, _ = self.search(user_client, collection_name, vectors_to_search) assert len(res) > 0 # revoke CollectionReadOnly — now denied self.revoke_privilege_v2(client, role_name, "CollectionReadOnly", collection_name=collection_name) time.sleep(10) user_client, _ = self.init_milvus_client(uri=uri, user=user_name, password=password) self.search(user_client, collection_name, vectors_to_search, check_task=CheckTasks.check_permission_deny) # cleanup self.drop_collection(client, collection_name)