《AI Agent 工程实战》4.2 授权与租户隔离：workspace 行级隔离

user ──< membership >── workspace ──< runs / sessions / memories / api_keys

# api/repo.py
class RunRepo:
    """所有查询强制带 workspace_id —— 租户边界写在 SQL 里，而不是靠调用方自觉。"""

    def __init__(self, conn):
        self.conn = conn

    def get(self, workspace_id: str, run_id: str):
        return self.conn.execute(
            "select id, workspace_id, question from runs "
            "where workspace_id = %s and id = %s",
            (workspace_id, run_id),
        ).fetchone()          # 不属于该租户 -> None，上层映射成 404

    def get_unsafe(self, run_id: str):
        # 反例：忘了 where workspace_id，只按主键取 —— 跨租户泄露
        return self.conn.execute(
            "select id, workspace_id, question from runs where id = %s",
            (run_id,),
        ).fetchone()

docker run --name ab-pg -e POSTGRES_PASSWORD=postgres \
  -e POSTGRES_DB=research -p 15435:5432 -d postgres:17-alpine

# demo_42.py
import psycopg
from psycopg.rows import dict_row

DSN = "postgresql://postgres:postgres@localhost:15435/research"

with psycopg.connect(DSN, row_factory=dict_row) as conn:
    conn.execute("create temp table runs_demo "
                 "(id text primary key, workspace_id text not null, question text)")
    conn.execute("insert into runs_demo values "
                 "('run_100','ws_acme','ResearchCopilot 是什么'),"
                 "('run_200','ws_globex','竞品分析')")
    repo = RunRepo(conn)

    def fetch(ws, run_id):
        row = repo.get(ws, run_id)
        if row is None:
            return 404, {"code": "run_not_found", "message": "运行不存在", "details": {}}
        return 200, row

    for label, ws, run_id in [
        ("A 租户取自己的 run", "ws_acme", "run_100"),
        ("B 租户取 A 的 run", "ws_globex", "run_100"),
        ("A 租户取不存在的 run", "ws_acme", "run_999"),
    ]:
        code, body = fetch(ws, run_id)
        print(f"{label:<24} {code}  {body}")

A 租户取自己的 run             200  {'id': 'run_100', 'workspace_id': 'ws_acme', 'question': 'ResearchCopilot 是什么'}
B 租户取 A 的 run            404  {'code': 'run_not_found', 'message': '运行不存在', 'details': {}}
A 租户取不存在的 run            404  {'code': 'run_not_found', 'message': '运行不存在', 'details': {}}
忘了 where 的反例           泄露 {'id': 'run_100', 'workspace_id': 'ws_acme', 'question': 'ResearchCopilot 是什么'}

row = repo.get_unsafe(run_id)          # 已经把数据取出来了
if row and row["workspace_id"] == ws:  # 判断在应用层
    return row
return 404

# demo_42_rls.py（节选）
with psycopg.connect(DSN) as conn:
    conn.execute("create table runs_rls "
                 "(id text, workspace_id text not null, question text)")
    conn.execute("insert into runs_rls values "
                 "('run_100','ws_acme','ResearchCopilot 是什么'),"
                 "('run_200','ws_globex','竞品分析')")
    conn.execute("create role rls_app nologin")
    conn.execute("grant usage on schema public to rls_app")
    conn.execute("grant select on runs_rls to rls_app")

    conn.execute("alter table runs_rls enable row level security")
    conn.execute("create policy ws_iso on runs_rls "
                 "using (workspace_id = current_setting('app.workspace_id', true))")

    conn.execute("set role rls_app")           # 扮演应用连接
    conn.execute("set app.workspace_id = 'ws_acme'")
    print("ws_acme 全表扫 ->", conn.execute(
        "select id, workspace_id from runs_rls order by id").fetchall())
    conn.execute("set app.workspace_id = 'ws_globex'")
    print("ws_globex 全表扫 ->", conn.execute(
        "select id, workspace_id from runs_rls order by id").fetchall())
    print("ws_globex 忘写 where 直取 run_100 ->", conn.execute(
        "select id, workspace_id from runs_rls where id = 'run_100'").fetchall())

ws_acme 全表扫 -> [('run_100', 'ws_acme')]
ws_globex 全表扫 -> [('run_200', 'ws_globex')]
ws_globex 忘写 where 直取 run_100 -> []