企业RAG构建中，如何用“行级别权限管控”避免数据泄露

01.

背景介绍

在现代数据管理中，如何高效地进行权限控制是企业面临的一个重要挑战。尤其在涉及多个部门和角色的场景下，确保数据的安全性和访问的便捷性变得尤为关键，基于此 Milvus 提供了 RBAC（Role-Based Access Control ）功能。然而，一些关键行业却对数据的权限异常敏感且有着实际的应用场景，譬如：

在医疗行业中，医院和医疗机构需要保护患者的隐私，同时又需要在专业人员之间共享必要的信息。例如，一位医生可能需要访问其患者的完整医疗记录，以提供准确的诊断和治疗方案。但这位医生不应有权限查看非其负责患者的医疗信息。通过行级别权限管理，向量数据库可以精确地控制每个数据行的访问权限，确保只有被授权的医疗人员才能访问特定患者的数据。这种精细的权限控制有助于遵守医疗行业的法规，满足 HIPAA 的合规要求。

金融行业同样需要严格的权限管理。银行和金融机构处理着客户的财务数据，包括账户信息、交易记录和信用评分等。这些数据被转换为向量，用于风险评估、欺诈检测和个性化服务。然而，如果这些敏感数据被内部人员滥用或泄露，可能会导致严重的财务损失和法律后果。通过行级别权限管理，金融机构可以确保只有特定的员工，例如负责某一客户的客户经理，才能访问对应的数据。风险控制部门可能需要查看更广泛的数据以进行分析，但他们的访问也应受到监控和限制，以防止数据滥用。

基于这种细粒度权限管控的需求，本篇文章将介绍一种基于角色和权限的控制机制，该机制采用位图索引来管理数据表中行级别的访问权限，使得权限控制更为精细化和高效。这种方法不仅能够高效处理大规模数据的权限查询，还能灵活应对权限的更新操作。文章将从角色与权限的定义、位图的构建与使用、以及如何在实际场景中实现和应用这些概念等方面进行详细阐述。同时，我们还将通过 Milvus 的具体功能演示，展示如何在企业知识管理系统中利用这一机制来实现部门级的精细化权限控制。

02.

基本原理

2.1 角色和权限的定义

• 角色（Role）: 定义用户在系统中的角色，每个角色对应一组权限。

• 权限（Permission）: 定义在数据表中对某些行的访问权限（如读取、写入、删除等）。

2.2 位图索引的构建

• 每个角色都会有一个位图来表示其可以访问的数据行。

• 位图的长度与数据表的行数相等，每一位代表一行，如果该位为 1，则表示该角色对该行有访问权限，为 0 则表示没有访问权限。

2.3 位图的使用

• 权限赋予: 当给一个角色赋予某行的权限时，将该角色的对应位图中该行的位置置为 1。

• 权限判断: 判断某个角色对某行是否有权限时，只需检查该行对应的位图位是否为 1。

2.4 示例

假设有一张表 Collection A 用于存放企业知识，不同的知识内容用 doc_id 标识，其所属的知识库用 kb_id 标识。

定义两个角色：

• Role 1：可以访问第 1、2、3、 4 行数据，即 kb_id = 1 的内容。

• Role 2：可以访问第 5 行数据，即 kb_id = 2 的内容。

2.5 查询操作

当用户需要查询某些数据时，可以通过角色的位图与查询条件组合来快速筛选出用户有权限访问的行。例如，用户属于 Role 1，当查询“所有数据”时，通过位图 11110 筛选出 Data A, Data B, Data C, 和 Data D。

2.6 权限更新

如果需要给某个角色添加或者删除某行的权限，只需对对应位图中的相应位进行更新即可。

2.7 优点与考虑

• 高效性: 位图操作非常高效，适合大规模数据的权限管理。

• 空间开销: 位图占用空间较小，尤其在行数较多时。

• 灵活性: 适用于多种查询条件的组合。

03.

Milvus具体功能演示

场景：企业 RAG，不同部门之间有不同的 Knowledge base，有些是公开的，有些是机密的，希望基于 entity/document 管理这些权限

Role 定义：假设角色中有超级管理员 admin, 以及下面根据业务部分组成的不同角色, 例如 [ceo, finance, sales, developer] 等

3.1 权限列定义

在上面所示的权限模型中，采用 array 列存储权限信息，field_name 可以自定义，array 的最大长度根据每个用户自己具体的使用场景设定，并在此列建立 BITMAP 索引，如下所示：

# 1. Set up a Milvus client
client = MilvusClient(
    uri=CLUSTER_ENDPOINT
)

# 2. Create a collection
schema = MilvusClient.create_schema(
    auto_id=False,
    enable_dynamic_field=False,
)

# 3. define schema 
schema.add_field(field_name="id", datatype=DataType.INT64, is_primary=True)
schema.add_field(field_name="data", datatype=DataType.VARCHAR, max_length=100)
schema.add_field(field_name="vector", datatype=DataType.FLOAT_VECTOR, dim=128)

# 4. add security column
schema.add_field(field_name="security_group", datatype=DataType.ARRAY, 
                 element_type=DataType.VARCHAR, max_capacity=10, max_length=100)

index_params = MilvusClient.prepare_index_params()
index_params.add_index(
    field_name="vector",
    index_type="IVF_FLAT",
    metric_type="L2",
    params={"nlist": 1024}
)

# 5. create bitmap index for security column
index_params.add_index(field_name="security_group", 
                       index_type="BITMAP")

# 6. create collection
client.create_collection(
    collection_name="test_collection",
    schema=schema,
    index_params=index_params
)

3.2 权限写入

在写入每行数据的时候需要指定该行数据可以被读取的role 有哪些，即在权限列写入role 角色即可， 如下所示

data =[]
data.append({
        "id": random.randint(0, 100000),
        "vector": [ random.uniform(-1, 1) for _ in range(128) ],
        "data": "data" + str(random.randint(0,100000)),
        # ceo role can read
        "security_group": ["ceo"]
})

data.append({
        "id": random.randint(0, 100000),
        "vector": [ random.uniform(-1, 1) for _ in range(128) ],
        "data": "data" + str(random.randint(0,100000)),
         # finance role can read
        "security_group": ["finance"]
})

data.append({
        "id": random.randint(0, 100000),
        "vector": [ random.uniform(-1, 1) for _ in range(128) ],
        "data": "data" + str(random.randint(0,100000)),
        # both sales and developer can read
        "security_group": ["sales", "finance"]
})

res = client.insert(collection_name="test_collection", data=data)

3.3 权限查询

当在search 或者query 查询时，需要加入对特定role 的限定查询，即只能看到指定role 的数据，其他数据对于查询role 不可见，如下所示：

3.3.1 只查询可见列

res = client.query(
    collection_name="test_collection",
    # 查询仅 ceo role 可见的数据
    filter='array_contains(security_group, "ceo")',
    output_fields=["id", "data", "security_group"],
)
print("ceo role read:")
print(res)

res = client.query(
    collection_name="test_collection",
    # 查询仅 sales role 可见的数据
    filter='array_contains(security_group, "sales")',
    output_fields=["id", "data", "security_group"],
)
print("sales role read:")
print(res)

res = client.query(
    collection_name="test_collection",
    # 查询仅 developer 可见的数据
    filter='array_contains(security_group, "develop")',
    output_fields=["id", "data", "security_group"],
)
print("developer role read:")
print(res)

res = client.query(
    collection_name="test_collection",
    # 查询仅 developer 或者 ceo 可见的数据
    filter='array_contains_any(security_group, ["develop", "ceo"])',
    output_fields=["id", "data", "security_group"],
)
print("developer or ceo role read:")
print(res)

以上查询结果如下所示：

ceo role read:
data: [
"{'security_group': ['ceo'], 'id': 3443, 'data': 'data35077'}", 
"{'security_group': ['ceo'], 'id': 12181, 'data': 'data99090'}", 
"{'security_group': ['ceo'], 'id': 16551, 'data': 'data74619'}", 
"{'security_group': ['ceo'], 'id': 24466, 'data': 'data1373'}", ...
sales role read:
data: [
"{'data': 'data75305', 'security_group': ['sales'], 'id': 9122}", 
"{'data': 'data61054', 'security_group': ['sales'], 'id': 20087}", 
"{'data': 'data47948', 'security_group': ['sales', 'develop'], 'id': 21726}", 
"{'data': 'data8596', 'security_group': ['sales'], 'id': 40090}", ... 
developer role read:
data: [
"{'data': 'data1515', 'security_group': ['develop'], 'id': 6429}", 
"{'data': 'data47031', 'security_group': ['develop'], 'id': 10953}", 
"{'data': 'data47948', 'security_group': ['sales', 'develop'], 'id': 21726}", 
"{'data': 'data86894', 'security_group': ['develop'], 'id': 56980}"], ... 
developer or ceo role read:
data: [
"{'data': 'data35077', 'security_group': ['ceo'], 'id': 3443}",
 "{'data': 'data1515', 'security_group': ['develop'], 'id': 6429}", 
 "{'data': 'data47031', 'security_group': ['develop'], 'id': 10953}", 
 "{'data': 'data99090', 'security_group': ['ceo'], 'id': 12181}", ...

可精准展示具体角色所能看到的数据，而屏蔽其权限之外的内容，同时可以通过 security_group 的 array 自由叠加权限。

3.3.2 自定义 filter + role 权限可见

用户某些情形也可以加一下自定义的查询 filter, 这些 filter 会 apply 在标量列的查询条件中，即在查询时，需加上role 的可见性与 filter 做联合查询，如下所示：

res = client.query(
    collection_name="test_collection",
    # sales role 查询 filter "pk in [1, 3, 5]"
    filter='pk in [1, 3, 5] && array_contains(security_group, "sales")',
    output_fields=["id", "data", "security_group"],
)

res = client.query(
    collection_name="test_collection",
    # developer role 查询 filter "pk > 10"
    filter='pk > 10 && array_contains(security_group, "develop")',
    output_fields=["id", "data", "security_group"],
)

3.4 权限更新

某些时候，需要对权限进行更改，如增加某行数据对于某个Role 的权限，或者删除某行数据对于Role 的可见性，

可通过milvus 的upsert 接口更新。

如下所示：

upsert_row_update = {
        "id": 101,
        "vector": upsert_vector,
        "data": upsert_data,
        # update role 
        "security_group": ["finance", "sales"]
}
res = client.upsert(
    collection_name="test_collection",
    data=upsert_row_update)

结果：

pk = 101:
data: [" {'id': 101,
'data': 'data63309', 
'vector': [0.38069534, 0.15088418, -0.6266929, -0.6038463, 0.2516377...],
'security_group': ['finance'],"]  

after upsert
data: ["{'id': 101, 
'data': 'data63309', 
'vector': [0.38069534, 0.15088418, -0.6266929, -0.6038463, 0.2516377...], 
'security_group': ['finance', 'sales']}"]  

04.

小结

通过 array 列定义的 security_group 和位图索引过滤模式，已经可以实现基本的行级别读权限管理能力，灵活应对查询时的权限管理需求。这种方法在高效性和细粒度权限控制方面表现出色，但也对管理员的操作提出了更高的要求，包括在插入或更新数据时进行细粒度的权限管理，以及在创建表时具备全局视角做好权限规划。在权限的动态管理和继承等易用性方面，Milvus 在云上的版本 Zilliz Cloud 也有计划进一步完善细粒度权限管理功能，以提高用户的管理效率和灵活性，满足更多样的业务需求。