存储、分发注册表与非旁路的多租户/ABAC 隔离

30 秒导读: 前几章讲的是"请求怎么进来、怎么路由、怎么编排"(见 请求生命周期、路由与适配、Responses 编排)。这一章讲数据落在哪、重启后怎么活下来、多个租户/用户的数据凭什么不会互相看见。核心精华只有一句:OGX 把"权限"做进了每一条 SQL 的 WHERE 子句——不是"应用层记得检查就检查",而是结构上绕不过去。

本章讲支撑全局的持久化底座和安全隔离底座。它不产生业务逻辑,但所有业务逻辑的数据都经过它。

---

1. 这是什么(零基础也能懂)

一句话定义: 这是 OGX 的"磁盘层"——两套存储抽象(键值 KVStore、关系 SqlStore),外加一层强制隔离的安全包装(AuthorizedSqlStore)。

解决什么问题: 一个 API 服务器要记住很多东西——

• 控制面数据: 你注册过哪些模型、向量库、工具组?(注册表,重启后要还在)
• 业务面数据: 每次推理的日志、对话历史、prompt、连接器配置。
• 谁能看谁的数据: 如果 OGX 被多个团队/公司共用,A 公司的对话历史绝不能被 B 公司读到。

给谁用: 平台方(自己搭 OGX 给内部多团队用)、SaaS 提供方(一套 OGX 服务多个客户)。单机自用的人可以完全无视隔离层——它默认关闭。

一句话直觉/类比: 把 KVStore 当一个持久化的字典(存"注册表"这种键值元数据),把 SqlStore 当一张张带列的表(存日志这种结构化记录)。而 AuthorizedSqlStore 就像给每张表焊了一道闸门:任何人进来查数据,系统都会自动、悄悄地在他的查询后面加上"…… AND 这行属于你这个租户 AND 你有权看它",他改不掉、绕不过。

用起来什么样: 业务代码从不直接碰数据库。它只声明"我要一个受权限保护的 SQL 存储",然后照常读写:

# 示意,非源码 —— 展示业务层视角有多"无感"
store = await authorized_sqlstore(reference, policy)   # 拿到受保护的表
await store.create_table("inference_store", {"id": ColumnType.STRING, ...})
await store.insert("inference_store", {"id": "req-1", "model": "gpt-4o"})  # 自动盖上 owner + tenant
rows = await store.fetch_all("inference_store")   # 自动只返回"我"能看的行

业务层完全不写 WHERE tenant_id = ...——隔离是存储层替它做的。这正是"非旁路(non-bypassable)"的含义。

---

2. 顶层全景(它大概怎么转)

怎么读这张图: 从上到下是"业务代码 → 抽象接口 → 安全包装 → 真实后端"。左边一列是控制面(注册表走 KVStore),右边一列是业务面(日志/对话走受保护的 SqlStore)。

        控制面(资源注册)                     业务面(运行数据)
  ┌───────────────────────────┐      ┌────────────────────────────────┐
  │ 路由表 / 资源注册          │      │ inference_store / conversations │
  │ (models, vector stores,    │      │  / prompts / connectors 等      │
  │  tool groups …)            │      │                                 │
  └─────────────┬─────────────┘      └────────────────┬───────────────┘
                │ register/get/get_all               │ insert/fetch_all/update
                ▼                                     ▼
  ┌───────────────────────────┐      ┌────────────────────────────────┐
  │  DistributionRegistry      │      │      AuthorizedSqlStore         │  ← 安全闸门
  │  (Cached / Disk)           │      │  ① 切租户 WHERE tenant_id=?     │
  │  带 5s TTL 内存缓存        │      │  ② 过 ABAC 策略(owner/attrs)   │
  └─────────────┬─────────────┘      └────────────────┬───────────────┘
                │                                     │ 剥掉安全字段后透传
                ▼                                     ▼
  ┌───────────────────────────┐      ┌────────────────────────────────┐
  │        KVStore             │      │           SqlStore              │
  │  get/set/values_in_range   │      │  create_table/insert/fetch_all  │
  └─────────────┬─────────────┘      └────────────────┬───────────────┘
                ▼                                     ▼
     SQLite / Redis / Postgres / MongoDB       SQLite / Postgres (SQLAlchemy)

部件一句话职责:

部件	干什么	在哪个文件
KVStore	键值接口(get/set/范围扫描),值是字符串(通常 JSON)	src/ogx/core/storage/kvstore/kvstore.py
SqlStore	带列定义的表操作(建表/增删改查/分页)	src/ogx_api/internal/sqlstore.py (Protocol)、sqlalchemy_sqlstore.py(实现)
DistributionRegistry	把 models/vector stores/tool groups 等资源持久化进 KVStore,支撑重启存活	src/ogx/core/store/registry.py
AuthorizedSqlStore	在每次读写上强制"先切租户、再过 ABAC"	src/ogx/core/storage/sqlstore/authorized_sqlstore.py
存储配置/引用	StorageBackendType、KVStoreReference、SqlStoreReference 等类型	src/ogx/core/storage/datatypes.py

主线走一遍(高层): 服务器启动时 _initialize_storage() 把 run config 里的 backends 拆成 kv/sql 两组注册进全局表,并把租户模式一次性盖到进程级(src/ogx/core/stack.py:718-736)。此后:控制面通过 DistributionRegistry 把资源写进 KVStore;业务面通过 authorized_sqlstore() 工厂拿到带安全闸门的表来读写。两条线共用底层的 SQLite/Postgres 等后端,但走的抽象和保护完全不同。

---

3. 底座之一:两套存储抽象

先看两套"接口",它们刻意保持简单,把复杂度留给上层。

3.1 KVStore —— 持久化的字典

它要解决的小问题: 有些数据天生是"键 → 一坨 JSON",不需要列结构,只需要按键读写、能按前缀范围扫描。典型就是分发注册表、配额中间件、skills 元数据。

接口长这样(五个方法):

方法	作用
get(key) / set(key, value, expiration)	单键读写,可带过期时间
delete(key)	删键
values_in_range(start, end) / keys_in_range(start, end)	按字典序范围扫描(注册表靠它一次拉全量)

真实接口见 src/ogx_api/internal/kvstore.py(Protocol),内存实现 InmemoryKVStoreImpl 在 src/ogx/core/storage/kvstore/kvstore.py:43。

四个后端,一个工厂。 kvstore_impl() 按配置类型分派到具体实现,并做按 (backend, namespace) 去重的单例缓存——同一个引用第二次要就直接返回已建实例:

# src/ogx/core/storage/kvstore/kvstore.py:131 kvstore_impl —— 节选
existing = _KVSTORE_INSTANCES.get(cache_key)
if existing:
    return existing            # 单例:同一 (backend, namespace) 不重复建
...
if isinstance(config, RedisKVStoreConfig):
    impl = RedisKVStoreImpl(config)
elif isinstance(config, SqliteKVStoreConfig):
    impl = SqliteKVStoreImpl(config)
# … Postgres / MongoDB 同理

支持的后端由 StorageBackendType 枚举定义(datatypes.py:20):kv_redis / kv_sqlite / kv_postgres / kv_mongodb。默认是 SQLite(_default_backends() 建 kvstore.db,datatypes.py:360)。

3.2 SqlStore —— 带列的表

它要解决的小问题: 业务数据是结构化的(推理日志有 model、tokens、时间…),需要建表、按列过滤、分页游标。这类走 SqlStore。

接口关键点: create_table / insert / upsert / fetch_all / update / delete / add_column_if_not_exists,见 src/ogx_api/internal/sqlstore.py:48 的 SqlStore Protocol。注意每个查询方法都留了一对 where_sql / where_sql_params 参数——这正是安全层注入过滤条件的钩子,后面 §4 会用到。

只有两个后端: SQLite、Postgres,都基于 SQLAlchemy(_sqlstore_impl() 只会实例化 SqlAlchemySqlStoreImpl,sqlstore.py:78)。为什么比 KVStore 少两个?因为业务表要跑真实 SQL(JSON 提取、ALTER TABLE 迁移),Redis/Mongo 不适配。

一个懒加载的巧思: create_table() 不立刻建表,只登记 SQLAlchemy 元数据,真正的 CREATE/ALTER 推迟到第一次数据操作时的 _ensure_engine()(sqlalchemy_sqlstore.py:181-213)。add_column_if_not_exists() 同理——引擎没起来就把列加进"待办队列"(_pending_columns),起来了才真去 ALTER TABLE(sqlalchemy_sqlstore.py:460-476)。这让 Stack.initialize() 能在临时事件循环里声明表结构,而真正连库发生在 uvicorn 的请求循环里(配合 reset_sqlstore_engines(),sqlstore.py:101)。

谁在用 SqlStore: inference store(推理日志)、conversations、prompts、connectors、responses——但它们都不直接用,而是隔着 §4 的安全层。

---

4. 底座之二:分发注册表(重启存活 + 刷新)

这节讲控制面。资源注册的路由/解析逻辑在 Provider 注册与解析 讲过,这里只讲它怎么落盘、怎么在多 worker 下保持一致。

它要解决的小问题: 你 register 了一个模型,服务器重启后它得还在;而且生产环境常是多进程(多 worker),进程 A 注册的资源,进程 B 也得看得见。

思路: 用 KVStore 做真源(persist),用一层带 TTL 的内存缓存做加速,并靠定期刷新解决多 worker 一致性。

键的设计。 每个资源按 type:identifier 编码成一个带版本前缀的 key,范围扫描时用一个 \xff 结尾界定上界:

# src/ogx/core/store/registry.py:40-48
REGISTER_PREFIX = "distributions:registry"
KEY_VERSION = "v10"
KEY_FORMAT = f"{REGISTER_PREFIX}:{KEY_VERSION}::" + "{type}:{identifier}"

def _get_registry_key_range() -> tuple[str, str]:
    start_key = f"{REGISTER_PREFIX}:{KEY_VERSION}"
    return start_key, f"{start_key}\xff"     # \xff 作为该前缀下的字典序上界

get_all() 就是对这个 range 做一次 values_in_range,再把每个 JSON 反序列化回 RoutableObjectWithProvider(registry.py:78、_parse_registry_values 在 :51)。

两个实现,继承关系。

类	特点	用途
DiskDistributionRegistry	纯 KVStore,无缓存,get_cached 直接抛 NotImplementedError	基类
CachedDiskDistributionRegistry	在前者之上加内存 dict 缓存 + 5s TTL 刷新 + 锁	生产默认(create_dist_registry 造的就是它,:273)

多 worker 一致性的两处细节(精华):

1. 读放行也回源。 get_all() 在返回缓存前会 _refresh_cache_from_db()——只要距上次刷新超过 TTL,就重新从 KVStore 拉全量,"这样才能看见其它 worker 建的对象"(registry.py:208-217)。
2. 注册时不信自己的缓存。 register() 用 super().get()(直接读 DB)而不是 get_cached() 判存在,注释点明:多 worker 下每个进程各有一份缓存,必须读权威的存储对象(registry.py:238-244)。

重复注册怎么办(重启幂等): register() 遇到已存在的对象,若完全相等就 no-op;若incoming 是 existing 的子集(每个显式设置的字段都匹配)也放行——这覆盖"重启时 config 里的对象缺了运行期才补上的 owner 字段"这种情况;只有真冲突(同字段不同值)才报错(registry.py:107-131)。

---

5. 核心精华:非旁路的多租户 / ABAC 隔离

这是本章的皇冠。前面都是"把数据存好",这节是"凭什么别人偷不到你的数据"。

5.1 为什么"非旁路"是关键词

常见的权限做法是"应用层记得在查询里加过滤"。问题是——只要有一个 handler 忘了加,就漏了。OGX 的选择是:业务层根本拿不到"不带过滤"的存储对象。

authorized_sqlstore() 是"唯一被支持的、给 API 用的 SQL 存储入口"(authorized_sqlstore.py:115-125 的 docstring 原话)。你拿到的永远是 AuthorizedSqlStore,它的每个读写方法都在内部拼好安全 WHERE 子句再下发——你没有机会跳过。

AuthorizedSqlStore 叠了两层独立强制,顺序固定:先切租户(§5.2),再过 ABAC(§5.3)。

5.2 第一层:租户隔离(tenant isolation)

三种模式(TenancyMode,datatypes.py:223):

模式	含义	tenant_id 列
disabled	默认,单机自用,不隔离	不加列
single	单租户,所有数据盖同一个 default_tenant_id	加列
multi	多租户,按请求携带的 tenant_id 分区	加列

建表时自动加列。 tenancy 一旦开启,create_table() 会给 schema 补上 tenant_id(以及下节的 access_attributes / owner_principal)三根安全列,并对已存在的表做 add_column_if_not_exists 迁移:

# src/ogx/core/storage/sqlstore/authorized_sqlstore.py:190-197 —— 节选
if self.tenancy_mode != TenancyMode.DISABLED and "tenant_id" not in enhanced_schema:
    enhanced_schema["tenant_id"] = ColumnType.STRING
...
if self.tenancy_mode != TenancyMode.DISABLED:
    await self.sql_store.add_column_if_not_exists(table, "tenant_id", ColumnType.STRING)

写入盖章。 任何 insert 都经 _enhance_item_with_access_control():先剥掉客户端自带的 tenant_id(绝不信任),再盖上当前认证用户的真实值(authorized_sqlstore.py:80-103)。注释直说 "Never trust client-supplied access control fields."

读改一律加 WHERE tenant_id=?。 这是整个隔离的命门——_build_tenant_filter():

# src/ogx/core/storage/sqlstore/authorized_sqlstore.py:225-233
def _build_tenant_filter(self, current_user: User | None) -> tuple[str, dict[str, Any]]:
    """Non-bypassable tenant partition filter. Applied before ABAC."""
    if self.tenancy_mode == TenancyMode.DISABLED:
        return "1=1", {}
    if not current_user or not current_user.tenant_id:
        if self.tenancy_mode == TenancyMode.SINGLE and self.default_tenant_id:
            return "tenant_id = :_tenant_id_filter", {"_tenant_id_filter": self.default_tenant_id}
        return "1=0", {}       # ← multi 模式缺租户上下文 = 默认拒绝,什么都看不到
    return "tenant_id = :_tenant_id_filter", {"_tenant_id_filter": current_user.tenant_id}

这段是精华中的精华。 三个分支要读懂:

• disabled → 1=1(恒真,不过滤)。
• multi 但没有租户上下文 → 1=0(恒假,默认拒绝)。哪怕代码里忘了鉴权,查询也返回空集——失败方向是"看不到",不是"全看到"。
• 正常有租户 → tenant_id = :param(参数化,防注入)。

这个 tenant 过滤在 fetch_all / update / delete / _check_access_for_rows 里都会被 _combine_where_clauses 与 ABAC 子句用 AND 拼在一起(authorized_sqlstore.py:343-360、:462-496)——且租户过滤永远在,ABAC 只是在其之上再收紧。

5.3 第二层:ABAC(基于属性的访问控制)

它要解决的小问题: 同一个租户内,还想区分"谁建的、谁能看"。比如"只有 owner 能读自己的对话"。

两根列 + 一套策略。 除 tenant_id 外,每张表还有 owner_principal(谁建的)和 access_attributes(JSON,如 {"roles": ["admin"], "teams": [...]})。策略是一组 AccessRule,默认策略被硬编码成 SQL_OPTIMIZED_POLICY(authorized_sqlstore.py:62-77),语义是:

• 无主记录(owner_principal = '')人人可读;
• 有主记录:用户是 owner,或 access_attributes 里任一类别(roles/teams/projects/…)与用户匹配,即放行。

两条执行路径(性能与正确性的权衡):

路径	何时用	怎么做
SQL 下推	用默认策略时	把策略翻译成 WHERE 子句(_build_default_policy_where_clause,:652),数据库层就过滤掉,不返回
保守 + 应用层复核	自定义策略时	SQL 只做"绝对该拒的"最小过滤(_build_conservative_where_clause,:686),取回后再逐行跑 is_action_allowed 复核(fetch_all 的 :373-390)

为什么要硬编码一份策略再做一致性校验? 因为 SQL 下推是"手写的 SQL 翻译",若上游改了 default_policy() 而 SQL 版没跟上,过滤就会错。所以构造时 _validate_sql_optimized_policy() 会比对二者,不一致就告警并退回保守模式(安全但慢)(authorized_sqlstore.py:167-180)。这是"宁可慢、不可漏"的防御式设计。

JSON 属性匹配是跨库的。 access_attributes 是 JSON 列,Postgres 用 @> jsonb 包含判断、SQLite 用 json_each 展开判断(_json_array_contains_value,:619),且 JSON path 先过 _VALID_JSON_PATH_RE 白名单校验防注入(:578)。

5.4 进程级设置:一次盖章,处处生效

租户模式不是每次调用传参,而是启动时盖到进程级全局。Stack.initialize() → _initialize_storage() 调 set_default_tenancy_config():

# src/ogx/core/stack.py:731-736 —— 节选
from ogx.core.storage.sqlstore.authorized_sqlstore import set_default_tenancy_config
register_kvstore_backends(kv_backends)
register_sqlstore_backends(sql_backends)
set_default_tenancy_config(run_config.server.tenancy)   # 进程级盖章

此后所有 authorized_sqlstore(reference, policy)(不显式传 tenancy_mode)都读这个进程级默认(authorized_sqlstore.py:35-49、:115-125)。好处:已有调用点一行不改就获得隔离能力——隔离是"配置驱动",不是"代码驱动"。

注:进程级 API 有两个入口——set_default_tenancy_config()(收完整 TenancyConfig,含 default_tenant_id)和 set_default_tenancy_mode()(只设 mode)。stack.py 走前者,后者是 mode-only 的便捷设置(authorized_sqlstore.py:35-49)。

5.5 一条 fetch 的完整走位(把两层串起来)

怎么读: 从上到下是一次 fetch_all 的内部步骤,①②是两道闸门。

业务调用 store.fetch_all("conversations")
        │
        ▼
 拿当前认证用户 get_authenticated_user()
        │
        ├─① 租户闸门 _build_tenant_filter()  →  "tenant_id = :t"  (缺上下文→"1=0")
        │
        ├─② ABAC 闸门 _build_access_control_where_clause() → "(owner='' OR owner=:me OR …)"
        │
        ▼
 _combine_where_clauses 用 AND 拼:  WHERE (ABAC) AND (tenant_id=:t)
        │
        ▼
 下发给底层 SqlStore.fetch_all(where_sql=…)  →  数据库只返回该租户+有权的行
        │
        ▼
 (自定义策略时)再逐行 is_action_allowed 复核 → 返回 filtered_rows

两道闸门都在存储层内部完成,业务层看不到、改不了——这就是"非旁路"的具体形态。

---

6. 巧妙之处(可借鉴的技术)

• 失败朝安全方向倒。 multi 模式缺租户上下文时返回 1=0 而非 1=1——默认拒绝而非默认放行。忘了鉴权只会"看不到数据",不会"泄露数据"(authorized_sqlstore.py:232)。
• 权限做进 WHERE,而非应用层 if。 隔离条件由存储层拼进每条 SQL,业务层无法跳过;拿不到"裸"存储对象(authorized_sqlstore.py:115 的"唯一入口")。
• 硬编码策略 + 运行时一致性校验。 SQL 下推版策略与 default_policy() 不一致就告警退回保守模式,防"翻译漂移"导致漏过滤(:167-180)。
• 写入永远重新盖章、剥客户端字段。 owner_principal/access_attributes/tenant_id 一律用认证值覆盖,绝不信任 payload(:80-103)。
• 注册表读放行也回源、注册不信自缓存。 用 TTL 刷新 + 读 DB 判存在,解决多 worker 缓存不一致(registry.py:208、:238)。
• 懒建表/懒加列。 表结构声明与真实 DDL 解耦,让初始化能在临时事件循环里跑,连库延后到请求循环(sqlalchemy_sqlstore.py:181、:460)。

可信度佐证(不展开): 仓库带 src/ogx/testing/api_recorder.py 的 record/replay 机制(RECORD / REPLAY / RECORD_IF_MISSING 三态,api_recorder.py:55-60),集成测试用它回放真实请求——存储与隔离行为都有可复现的测试兜底。

---

7. 边界与局限(诚实)

• AuthorizedSqlStore 只支持 SQLite 与 Postgres。 构造时 _detect_database_type() 显式拒绝其它类型(authorized_sqlstore.py:158-165)——因为 JSON 属性匹配和 SQL 下推是逐库手写的。KVStore 那四个后端不享受这层隔离。
• 隔离只作用于 SqlStore 数据面。 分发注册表走 KVStore,其访问控制靠资源自带的 owner/access_attributes 字段(ResourceWithOwner)和上层路由表处理,不经 AuthorizedSqlStore 的两层闸门。
• disabled 是默认。 单机部署零隔离——多租户是显式 opt-in,平台方必须在 server.tenancy 里配置才生效。
• 自定义策略走保守 + 应用层复核,牺牲性能。 只有默认策略能全量 SQL 下推;自定义策略要把候选行取回内存再逐行判,大表上更重(:686、fetch_all 复核循环)。

---

8. 代码地图(导航索引)

主题	文件路径	关键符号
KVStore 接口/工厂/内存实现	src/ogx/core/storage/kvstore/kvstore.py	kvstore_impl、InmemoryKVStoreImpl、register_kvstore_backends
SqlStore 协议(接口)	src/ogx_api/internal/sqlstore.py	SqlStore、ColumnType、DeleteOperation
SqlStore SQLAlchemy 实现	src/ogx/core/storage/sqlstore/sqlalchemy_sqlstore.py	SqlAlchemySqlStoreImpl、create_table、add_column_if_not_exists
SqlStore 工厂/后端注册	src/ogx/core/storage/sqlstore/sqlstore.py	_sqlstore_impl、register_sqlstore_backends、reset_sqlstore_engines
分发注册表	src/ogx/core/store/registry.py	DistributionRegistry、CachedDiskDistributionRegistry、create_dist_registry、KEY_FORMAT
安全层(核心)	src/ogx/core/storage/sqlstore/authorized_sqlstore.py	AuthorizedSqlStore、authorized_sqlstore、_build_tenant_filter、_build_default_policy_where_clause、_enhance_item_with_access_control、SQL_OPTIMIZED_POLICY、set_default_tenancy_config、set_default_tenancy_mode
存储配置/引用类型	src/ogx/core/storage/datatypes.py	StorageBackendType、KVStoreReference、SqlStoreReference、StorageConfig、_default_backends
租户/用户数据类型	src/ogx/core/datatypes.py	TenancyMode、TenancyConfig、User、ResourceWithOwner
进程级盖章	src/ogx/core/stack.py	_initialize_storage(:718)、Stack.initialize
可信度佐证	src/ogx/testing/api_recorder.py	InferenceMode(record/replay)

相邻章节: 资源如何被注册与解析见 Provider 注册与解析;这些存储服务谁在消费(inference store、conversations、responses)见 请求生命周期 与 Responses 编排。