70 lines
3.3 KiB
Python
70 lines
3.3 KiB
Python
|
"""档位推导（「主体能看哪些档位」）的**唯一落点**（`G-03`）。
|
|||
|
|
|
|||
|
|
## 为什么单独成文件
|
|||
|
|
|
|||
|
|
档位规则此前内联在 `app/core/knowledge_contracts.py` —— 契约模块同时承担了
|
|||
|
|
「请求/响应形状」与「权限推导规则」两种职责。`D2.1` 批次 G 的 `G-02`（身份轴，方案甲）
|
|||
|
|
落地时要把「访客」从 RBAC 角色轴迁到身份轴，那次改动**只需要动本文件**，
|
|||
|
|
不必再动契约、检索服务与工具层。
|
|||
|
|
|
|||
|
|
## 硬约束
|
|||
|
|
|
|||
|
|
1. **档位只由鉴权结果推导**，不由查询内容、请求参数或工具层自判身份字段决定。
|
|||
|
|
2. **失败关闭**：认不出任何角色 ⇒ `{public}`（不是全集）；空集合同样收敛为 `{public}`。
|
|||
|
|
3. **`visibility` 是 Milvus 分区键** ⇒ 过滤表达式必须能下推为分区裁剪，故用 `in [...]`。
|
|||
|
|
4. 取值域只有 `public` / `registered`。`internal` **不是**本系统的档位：
|
|||
|
|
金标 `M-8`（档位越权）会把它计为越权，因此不要往取值域里加。
|
|||
|
|
"""
|
|||
|
|
|
|||
|
|
from __future__ import annotations
|
|||
|
|
|
|||
|
|
from collections.abc import Iterable
|
|||
|
|
|
|||
|
|
#: 可见性档位取值域（与 Milvus 集合的 `visibility` 字段、分区键同源）。
|
|||
|
|
PUBLIC_TIER = "public"
|
|||
|
|
REGISTERED_TIER = "registered"
|
|||
|
|
#: 全集；`visibility_expression()` 只在请求档位**恰好等于全集**时返回 None（即不过滤）。
|
|||
|
|
ALL_TIERS: frozenset[str] = frozenset({PUBLIC_TIER, REGISTERED_TIER})
|
|||
|
|
|
|||
|
|
#: 主体类型 → 允许的档位集合（`D2.4` §5.4 的 `TIERS_BY_SUBJECT`）。
|
|||
|
|
#:
|
|||
|
|
#: ⚠️ 表内**只有** `visitor` 与 `customer`。`advisor` / `operator` / `admin` 不在表内，
|
|||
|
|
#: 按 `tiers_for_roles` 的收敛规则只拿到 `{public}` —— 要放宽它们属**权限决策**，
|
|||
|
|
#: 须单独裁定，不在此静默扩张。
|
|||
|
|
TIERS_BY_SUBJECT: dict[str, frozenset[str]] = {
|
|||
|
|
"visitor": frozenset({PUBLIC_TIER}),
|
|||
|
|
"customer": ALL_TIERS,
|
|||
|
|
}
|
|||
|
|
|
|||
|
|
#: 主体无法识别时的收敛档位。**是 `{public}` 而非 `ALL_TIERS`** —— fail-closed。
|
|||
|
|
DEFAULT_TIERS: frozenset[str] = frozenset({PUBLIC_TIER})
|
|||
|
|
|
|||
|
|
|
|||
|
|
def tiers_for_roles(roles: Iterable[str]) -> frozenset[str]:
|
|||
|
|
"""把 `RequestContext.roles` 映射成「主体被允许的档位集合」。
|
|||
|
|
|
|||
|
|
多角色取**并集**（`customer` 已含 `public`，故并集即该主体的最大可见范围）；
|
|||
|
|
一个角色都识别不出时收敛为 `DEFAULT_TIERS` —— fail-closed，不是放行全集。
|
|||
|
|
"""
|
|||
|
|
allowed: frozenset[str] = frozenset()
|
|||
|
|
for role in roles:
|
|||
|
|
allowed |= TIERS_BY_SUBJECT.get(role, frozenset())
|
|||
|
|
return allowed or DEFAULT_TIERS
|
|||
|
|
|
|||
|
|
|
|||
|
|
def visibility_expression(tiers: frozenset[str]) -> str | None:
|
|||
|
|
"""档位集合 → Milvus 过滤表达式；返回 `None` 表示**无需过滤**。
|
|||
|
|
|
|||
|
|
只有档位**恰好等于 `ALL_TIERS`** 时才返回 `None`；其余情况（含空集合 —— 收敛为
|
|||
|
|
`DEFAULT_TIERS`）一律显式列出，不做「空集合 = 不过滤」这种 fail-open。
|
|||
|
|
|
|||
|
|
用 `in [...]` 而不是 `== "public"`：`visibility` 是**分区键**，Milvus 在分区键模式下
|
|||
|
|
只支持可下推为分区裁剪的表达式；`in` 同时支持多档位，正好对应「客户 = public +
|
|||
|
|
registered」。
|
|||
|
|
"""
|
|||
|
|
normalized = frozenset(tiers) or DEFAULT_TIERS
|
|||
|
|
if normalized == ALL_TIERS:
|
|||
|
|
return None
|
|||
|
|
listed = ", ".join(f'"{tier}"' for tier in sorted(normalized))
|
|||
|
|
return f"visibility in [{listed}]"
|