应用 MCP 的权限边界：从登录用户到数据库行隔离

林把工单助手接到海港团队，梅在山脊团队使用同一个助手。两人都通过登录，也都能调用名为 list_tickets 的工具。假如服务端只检查“这是已登录用户”，梅就可能拿到海港的工单。再给工具加一个 tenant_id 参数也解决不了：这个参数可以由调用方填写。

本文做一个可重复的授权实验。我们先约定每个用户、每种客户端能做什么，再用原创工单数据验证允许和拒绝的边界。程序仅使用 Python 标准库与 SQLite，模拟授权谓词；它不实现 MCP 传输，不验证 JWT，也不运行 PostgreSQL RLS。Supabase 部分用于说明真实系统的接入位置和下一轮验收方法。

你需要会运行 Python 3.10 或更新版本。把下面三个文件放进同一目录，执行 python -m unittest -v test_authorization.py 即可。运行过程只创建内存数据库，不联网，不读取账号，不部署服务。本次使用 Python 3.12.14、SQLite 3.53.1，离线运行的 18 个测试方法全部通过，部分方法包含多组矩阵用例。“通过”只适用于本文列明的模拟行为。

已有的 MCP 服务教程适合先理解工具接口，Codex MCP 配置处理客户端连接，Agent Bus讨论任务动作。本篇从多个真实用户共享应用入口这一层开始，不重复这些基础。

sub 回答当前请求代表谁，client_id 回答哪一个 OAuth 客户端获得了令牌。tenant_id 指向业务数据的归属；成员表描述用户现在是否属于这个租户。read、insert、update 则是本次操作。四者要在同一个决策里相遇，任何一个都不能独自表达全部权限。

这里的 client_id 是授权服务器签发并经过验证的声明，不是工具参数，也不是客户端随手提供的请求头。客户端能够持有有效令牌，只说明它通过了认证入口；它能否代表该用户修改海港工单，还要继续检查。租户 ID 即使不可猜测，也不充当密码。

我们的业务规则很小：report 只能读取；editor 可以读取、新建和更新。成员可以访问所属租户的全部工单，包括同租户他人提交的工单。缺失 client_id 的产品会话与未知客户端在这个 MCP 入口都拒绝。产品自己的网页若需要接受普通会话，应另定明确的入口规则，不能把 null 自动解释成更高权限。

工具提示和只读标注可以帮助客户端安排调用，但服务端必须独立拒绝越权动作。即使客户端绕过界面直接发送写请求，也应得到相同结果。

在 Supabase 的应用 MCP 方案中，发现层告诉客户端去哪里获得授权；认证层验证用户令牌，并把用户作用域的数据库客户端交给工具。发现请求本身尚未登录，因此要能到达发现层。部署在 Edge Functions 时，官方示例把网关 verify_jwt 设为 false，再由函数内的认证中间件执行验证。只复制前半个开关会留下未保护入口。

组合关系可读作：withOAuthProtectedResource 在外，withSupabase({ auth: 'user' }) 在内，最里面才是 MCP 处理器。工具使用请求上下文中的 supabase；不要把 supabaseAdmin 传入普通工具的闭包。本文不提供未经编译的完整部署程序，实际接线请依据 官方部署指南与 组件说明锁定版本后实现。

外部客户端还需要项目开启 OAuth 服务、配置客户端注册，并由应用提供同意界面。这套库的用户认证要求 ES256 或 RS256 签名，拒绝旧 HS256 用户令牌；这是该库的要求，不能推广成所有 OAuth 实现的通则。

版本核对日期为 2026-10-07。部署示例使用 server@1、middleware@1，两种组合方式要求 server 至少 1.6.0；CLI 2.117.0 起支持本地函数 slug 注入，影响发现地址的生成。仓库 README 把 server v1.X 标为 Public Beta。另一份 Library MCP 页面仍把 pipeline 组合标为 alpha、提到 middleware 0.x，并建议嵌套形式。本文只解释上述嵌套顺序，不把相互不同步的标签合并成整个产品的统一状态，也没有验证任何安装组合。

两个租户是海港 harbor 和山脊 ridge。林 lin 只属于海港，梅 mei 只属于山脊，乔 qiao 同时属于两者。H1、H2 属于海港；R1、R2 属于山脊。工单的 internal_note 是合成内部备注，不应返回给工具调用方。

expected.json 人工写下预期结果，不从程序的 member 函数生成。这样即使实现把租户关系写反，测试仍有机会发现。林查询全部应得到 H1、H2；林主动选择 ridge 应得到空列表；乔查询全部可得到四条。这同时防止一种过度修复：为了阻止跨租户访问，错误地把合法双租户成员也锁在单一租户。

写入矩阵另列旧行和目标租户。林可以在海港新建和修改 H1；林不能在山脊新建，也不能把 H1 移到山脊；林不能把原本无权读取的 R1 拉进海港。我们明确允许乔在两个已有成员资格的租户间移动工单。如果你的产品禁止任何跨租户移动，应增加“新旧 tenant_id 相等”的业务约束，并把乔的这一项预期改为拒绝。

expected.json

{
  "description": "Manually authored expected outcomes, independent of implementation predicates",
  "reads": [
    [
      "lin",
      "report",
      null,
      [
        "H1",
        "H2"
      ]
    ],
    [
      "lin",
      "editor",
      "ridge",
      []
    ],
    [
      "mei",
      "report",
      null,
      [
        "R1",
        "R2"
      ]
    ],
    [
      "qiao",
      "report",
      null,
      [
        "H1",
        "H2",
        "R1",
        "R2"
      ]
    ],
    [
      "qiao",
      "editor",
      "ridge",
      [
        "R1",
        "R2"
      ]
    ]
  ],
  "denied_clients": [
    null,
    "unknown"
  ],
  "writes": [
    [
      "lin",
      "report",
      "insert",
      null,
      "harbor",
      "deny"
    ],
    [
      "lin",
      "editor",
      "insert",
      null,
      "harbor",
      "allow"
    ],
    [
      "lin",
      "editor",
      "insert",
      null,
      "ridge",
      "deny"
    ],
    [
      "lin",
      "report",
      "update",
      "H1",
      "harbor",
      "deny"
    ],
    [
      "lin",
      "editor",
      "update",
      "H1",
      "harbor",
      "allow"
    ],
    [
      "lin",
      "editor",
      "update",
      "H1",
      "ridge",
      "deny"
    ],
    [
      "lin",
      "editor",
      "update",
      "R1",
      "harbor",
      "deny"
    ],
    [
      "qiao",
      "editor",
      "update",
      "H1",
      "ridge",
      "allow"
    ]
  ],
  "public_fields": [
    "id",
    "tenant_id",
    "title",
    "status"
  ]
}

Principal 是测试夹具提供的可信主体。它不是认证器：任何能运行这份 Python 的人都能构造 Principal，这正说明它不能直接放到互联网边界。真实系统必须先验证令牌的签名、签发者、目标受众、有效期等，再从验证结果建立主体；不能先解码一个字典就认定登录成功。

gate 先检查用户存在以及客户端动作。read 将主体的 sub 绑定进 SQL，关联当前成员表，再附加调用者选择的租户条件。选择 ridge 只能进一步缩小结果，不能扩大权限。参数绑定也避免把 tenant_id 当 SQL 代码执行。

insert 检查新行租户。update 先读旧行并检查旧归属，再检查新归属，任一失败便不修改数据。不存在与不可访问的旧行都返回同一个拒绝原因，减少通过这一更新入口的错误差异探测记录的机会。它不保证全面防枚举：本例 insert 仍由调用方给出全局唯一 ID，林用 R1 作为新工单 ID 会遇到主键冲突，即使他读不到山脊的 R1。生产入口应由服务端生成不可预测的 ID，限制标识探测，并脱敏冲突错误。本例省略标题长度、分页、审计、并发事务和完整业务验证；其目标是把授权条件做成看得清的程序。

SQLite 在这里存储夹具并执行显式 WHERE 条件；它不会替所有查询自动加上安全屏障。bad_ 开头的函数故意绕过条件，专门用于展示错误，不能接到真实工具。

authorization_lab.py

"""Offline predicate model. NOT JWT verification, MCP transport or PostgreSQL RLS."""
from dataclasses import dataclass
import sqlite3

PUBLIC_FIELDS = ("id", "tenant_id", "title", "status")
CLIENT_ACTIONS = {"report": {"read"}, "editor": {"read", "insert", "update"}}

@dataclass(frozen=True)
class Principal:
    # Trusted fixture ONLY: a real server constructs this after token validation.
    sub: str | None
    client_id: str | None

class Denied(Exception):
    pass

class Lab:
    def __init__(self):
        self.db = sqlite3.connect(":memory:")
        self.db.row_factory = sqlite3.Row
        self.db.executescript("""
        CREATE TABLE users(id TEXT PRIMARY KEY);
        CREATE TABLE tenants(id TEXT PRIMARY KEY);
        CREATE TABLE memberships(user_id TEXT, tenant_id TEXT,
          PRIMARY KEY(user_id, tenant_id));
        CREATE TABLE tickets(id TEXT PRIMARY KEY, tenant_id TEXT NOT NULL,
          title TEXT NOT NULL, status TEXT NOT NULL, internal_note TEXT NOT NULL);
        INSERT INTO users VALUES ('lin'), ('mei'), ('qiao');
        INSERT INTO tenants VALUES ('harbor'), ('ridge');
        INSERT INTO memberships VALUES
          ('lin','harbor'), ('mei','ridge'), ('qiao','harbor'), ('qiao','ridge');
        INSERT INTO tickets VALUES
          ('H1','harbor','Export stalls','open','synthetic: retry budget 3'),
          ('H2','harbor','Invoice label','closed','synthetic: manual credit review'),
          ('R1','ridge','Upload timeout','open','synthetic: private routing note'),
          ('R2','ridge','Search index','open','synthetic: incident draft');
        """)
        self.authz_epoch = 0
        self.cache = {}

    def gate(self, p, action):
        # Missing/unknown clients are denied on THIS MCP endpoint.
        if not p.sub or action not in CLIENT_ACTIONS.get(p.client_id, set()):
            raise Denied("identity or client action denied")
        if not self.db.execute("SELECT 1 FROM users WHERE id=?", (p.sub,)).fetchone():
            raise Denied("unknown user")

    def member(self, p, tenant):
        return self.db.execute(
            "SELECT 1 FROM memberships WHERE user_id=? AND tenant_id=?",
            (p.sub, tenant)).fetchone() is not None

    def raw(self, ticket_id):
        return self.db.execute("SELECT * FROM tickets WHERE id=?", (ticket_id,)).fetchone()

    def read(self, p, tenant=None):
        self.gate(p, "read")
        # SQLite executes an explicit predicate. It has no RLS here.
        rows = self.db.execute("""
          SELECT t.id,t.tenant_id,t.title,t.status FROM tickets t
          WHERE EXISTS (SELECT 1 FROM memberships m
                        WHERE m.user_id=? AND m.tenant_id=t.tenant_id)
          AND (? IS NULL OR t.tenant_id=?) ORDER BY t.id
        """, (p.sub, tenant, tenant))
        return [dict(r) for r in rows]

    def tool_read(self, p, args):
        # Reject identity-looking arguments instead of silently trusting them.
        if set(args) - {"tenant_id"}:
            raise Denied("unexpected argument")
        return self.read(p, args.get("tenant_id"))

    def insert(self, p, ticket_id, tenant, title):
        self.gate(p, "insert")
        if not self.member(p, tenant):
            raise Denied("new row tenant denied")
        with self.db:
            self.db.execute("INSERT INTO tickets VALUES (?,?,?,'open','')",
                            (ticket_id, tenant, title))
        self.authz_epoch += 1

    def update(self, p, ticket_id, new_tenant, title):
        self.gate(p, "update")
        with self.db:
            old = self.raw(ticket_id)
            if old is None or not self.member(p, old["tenant_id"]):
                raise Denied("old row denied")
            if not self.member(p, new_tenant):
                raise Denied("new row tenant denied")
            self.db.execute("UPDATE tickets SET tenant_id=?,title=? WHERE id=?",
                            (new_tenant, title, ticket_id))
        self.authz_epoch += 1

    def revoke(self, user, tenant):
        # Fixture administration; NOT exposed as an MCP tool.
        with self.db:
            self.db.execute("DELETE FROM memberships WHERE user_id=? AND tenant_id=?",
                            (user, tenant))
        self.authz_epoch += 1

    def cached_read(self, p, tenant=None):
        self.gate(p, "read")
        # Global epoch is deliberately simple; all mutations must invalidate it.
        key = (p.sub, p.client_id, tenant, "read", self.authz_epoch)
        if key not in self.cache:
            self.cache[key] = self.read(p, tenant)
        return [dict(r) for r in self.cache[key]]

# Intentionally broken examples. Never wire these into an application.
def bad_parameter_identity(lab, args):
    return lab.read(Principal(args["user_id"], args["client_id"]), args.get("tenant_id"))

def bad_client_only(lab, p):
    lab.gate(p, "read")
    return [dict(r) for r in lab.db.execute("SELECT * FROM tickets ORDER BY id")]

def bad_permissive_or(lab, p):
    lab.gate(p, "read")
    return [dict(r) for r in lab.db.execute("SELECT * FROM tickets ORDER BY id")
            if lab.member(p, r["tenant_id"]) or p.client_id == "report"]

def bad_admin_path(lab):
    return [dict(r) for r in lab.db.execute("SELECT * FROM tickets ORDER BY id")]

def bad_old_row_only(lab, p, ticket_id, tenant):
    lab.gate(p, "update")
    row = lab.raw(ticket_id)
    if not row or not lab.member(p, row["tenant_id"]):
        raise Denied("old row denied")
    with lab.db:
        lab.db.execute("UPDATE tickets SET tenant_id=? WHERE id=?", (tenant, ticket_id))

def bad_cache(lab, p, tenant=None):
    # Missing subject, client and membership version.
    if tenant not in lab.cache:
        lab.cache[tenant] = lab.read(p, tenant)
    return lab.cache[tenant]

def ids(rows):
    return [r["id"] for r in rows]

先看 test_05。调用方发送 user_id=mei、client_id=editor、tenant_id=ridge。坏函数完全相信这些参数，返回 R1、R2；安全入口拿到的主体仍是林，发现额外身份字段就拒绝。即使删除伪造字段，仅保留 tenant_id=ridge，也只能得到空列表。由此可以分辨“参数格式不允许”和“合法参数未命中有权数据”两种结果。

test_06 模拟只看客户端是否在白名单：report 的确是允许读的客户端，但省略成员条件后，林能读到四条工单。正确谓词必须把客户端动作与用户成员资格一起满足，结果才回到 H1、H2。

写入矩阵让林用 report 新建海港工单、更新 H1，两次都拒绝且数据保持原样；同一个林换成 editor 后，同租户操作允许。只读权限应在真实写入口落实，不能只让工具名称叫“只读”，或期望客户端遵守描述。

缺失与未知 client_id 则分别验证 read、insert、update 全部被 gate 拒绝。常见错误是“没识别到 OAuth 客户端，就当成第一方网页用户放行”。本文没有这个回退分支。空 sub、空字符串与不存在的用户同样拒绝，不能让缺失身份落入 SQL 或应用的宽松默认值。

test_07 故意只检查旧行：林拥有海港成员资格，H1 的旧行检查通过，坏函数便把它移到山脊。再次查询时林只剩 H2，梅却多了一条原属海港的工单。这个错误包含一次未经授权的数据转移，不只是显示结果少了一条。安全 update 在检查目标山脊时拒绝，原 H1 的归属保持海港。

反方向也要测。只检查新行会让林把 R1 改到海港，从而把原本看不见的记录拉进自己的范围。因此矩阵同时覆盖 H1→ridge 与 R1→harbor。读得见旧行、能在目标租户写入，两项都必须成立。

test_08 先让乔读取山脊，再删除乔的山脊成员资格。同一个 Principal 再次查询得到空列表，新建也被拒绝。成员资格来自数据库当前关系，未复制到长期使用的主体对象中。本实验是单进程顺序执行；它没有证明真实部署下正在进行的事务、令牌撤销传播或多个服务缓存会立即同步。

生产产品要先定义撤销语义：撤销后新请求应何时失效，正在运行的导出如何处理，已经交付给客户端的内容是否还可留存。收回成员资格不能让对方忘记此前已合法获得的资料。

test_09 把“是当前租户成员”与“客户端是 report”用 OR 相连。林只要带 report 身份，就对每条工单满足右侧条件，得到四条记录。正确版本把两项共同作为准入条件，返回 H1、H2。这个 Python 反例只验证布尔逻辑；真正 PostgreSQL 中多条 permissive 策略通常按 OR 组合，还要检查策略适用角色、命令和 restrictive 策略，不能靠新增一条严格策略就假设旧宽松策略失效。

test_10 的管理员路径直接读取全表。梅经过正常入口只得到 R1、R2，走该路径却得到四条以及内部备注。真实系统若使用绕过 RLS 的数据库身份，用户令牌验证得再认真，也不能替代行隔离。后台维护可以有独立的高权限路径，但普通 MCP 工具不应拿到该连接；日志、审批和凭据也应隔离。

test_11 让林先查询全部，坏缓存用 tenant=None 作为唯一键；梅的下一次查询不再经过数据库，直接复用 H1、H2。数据库谓词正确也救不了这次泄漏。本例安全缓存的键包含 sub、client_id、租户筛选、动作与全局 epoch，读取缓存之前还会执行 gate。

test_12 进一步检查撤销：乔的山脊结果已缓存，撤销成员后 epoch 递增，下一次查询重新计算为空。这个全局计数只适合内存演示。多实例应用需要可共享的权限版本或可靠失效机制；若有人绕过 revoke 直接改成员表，本例缓存就不会自动知道。因此不要把这个计数器照搬成分布式撤销方案。

test_13 对每条正常结果比较精确字段集合：只能有 id、tenant_id、title、status。internal_note 不应出现，哪怕调用方对该行有读取权。行权限不等于所有列都适合进入模型上下文。

在此模型中，SELECT 已显式列出四列；没有先 SELECT * 再盼望后续步骤记得删字段。如果应用允许直接数据库 API 访问，还须在数据库权限或受控视图等层面实现对应列限制，单靠 MCP 的投影无法限制其他入口。

权限拒绝、数据库错误和空结果也不应混在一起。示例对合法读取但没有可访问记录返回空列表，对未知主体或不允许的动作抛出 Denied。真实 MCP 适配器需要将它们转成清晰且脱敏的工具结果，并避免把 SQL、内部备注或令牌写进返回值和日志。

保存完整文件后，在该目录运行下面命令。测试代码会为大多数场景重建干净的内存数据库，写入矩阵的每一项也独立初始化，避免上一项操作污染下一项结果。故意错误的函数得到泄漏或非法移动，恰好是反例测试的预期；测试绿灯不表示坏函数安全。

python -m unittest -v test_authorization.py

成功时结尾是 Ran 18 tests 和 OK。不要只看这个数字：独立矩阵验证原始业务承诺；失败注入说明哪个条件缺失；拒绝操作还比较修改前后的数据，避免“报了错却已写入”。最后的几个测试覆盖租户参数绑定、未知动作、合法双租户移动与错误信息一致性。

test_authorization.py

"""Run: python -m unittest -v test_authorization.py"""
import json
from pathlib import Path
import unittest
from authorization_lab import (Lab, Principal, Denied, ids,
    bad_parameter_identity, bad_client_only, bad_permissive_or,
    bad_admin_path, bad_old_row_only, bad_cache)

EXPECTED = json.loads(Path(__file__).with_name("expected.json").read_text())
LIN = Principal("lin", "report")
EDIT = Principal("lin", "editor")
MEI = Principal("mei", "report")
QIAO = Principal("qiao", "editor")

class AuthorizationTest(unittest.TestCase):
    def setUp(self):
        self.lab = Lab()
    def tearDown(self):
        self.lab.db.close()
    def test_01_independent_read_matrix(self):
        for user, client, tenant, expected in EXPECTED["reads"]:
            with self.subTest(user=user, client=client, tenant=tenant):
                self.assertEqual(ids(self.lab.read(Principal(user, client), tenant)), expected)
    def test_02_independent_write_matrix(self):
        for user, client, action, row, tenant, result in EXPECTED["writes"]:
            lab = Lab()
            try:
                p = Principal(user, client)
                before = [tuple(r) for r in lab.db.execute("SELECT * FROM tickets ORDER BY id")]
                fn = (lambda: lab.insert(p, "N1", tenant, "New")) if action == "insert" else (
                    lambda: lab.update(p, row, tenant, "Changed"))
                with self.subTest(user=user, action=action, row=row, tenant=tenant):
                    if result == "deny":
                        with self.assertRaises(Denied): fn()
                        self.assertEqual(before, [tuple(r) for r in lab.db.execute("SELECT * FROM tickets ORDER BY id")])
                    else:
                        fn()
                        saved = lab.raw("N1" if action == "insert" else row)
                        self.assertEqual(saved["tenant_id"], tenant)
                        self.assertEqual(saved["title"], "New" if action == "insert" else "Changed")
            finally:
                lab.db.close()
    def test_03_missing_unknown_client_denied(self):
        for client in EXPECTED["denied_clients"]:
            for action in ("read", "insert", "update"):
                with self.subTest(client=client, action=action):
                    with self.assertRaises(Denied): self.lab.gate(Principal("lin", client), action)
    def test_04_empty_unknown_user_denied(self):
        for user in (None, "", "nobody"):
            with self.assertRaises(Denied): self.lab.read(Principal(user, "report"))
    def test_05_forged_identity_arguments(self):
        args = {"user_id":"mei", "client_id":"editor", "tenant_id":"ridge"}
        self.assertEqual(ids(bad_parameter_identity(self.lab, args)), ["R1", "R2"])
        with self.assertRaises(Denied): self.lab.tool_read(LIN, args)
        self.assertEqual(self.lab.tool_read(LIN, {"tenant_id":"ridge"}), [])
    def test_06_client_only_leaks(self):
        self.assertEqual(ids(bad_client_only(self.lab, LIN)), ["H1", "H2", "R1", "R2"])
        self.assertEqual(ids(self.lab.read(LIN)), ["H1", "H2"])
    def test_07_old_only_update_moves_unauthorized(self):
        bad_old_row_only(self.lab, EDIT, "H1", "ridge")
        self.assertEqual(self.lab.raw("H1")["tenant_id"], "ridge")
        self.assertEqual(ids(self.lab.read(LIN)), ["H2"])
        with LabContext() as clean:
            with self.assertRaises(Denied): clean.update(EDIT, "H1", "ridge", "Moved")
            self.assertEqual(clean.raw("H1")["tenant_id"], "harbor")
    def test_08_revocation_changes_next_query(self):
        self.assertEqual(ids(self.lab.read(QIAO, "ridge")), ["R1", "R2"])
        self.lab.revoke("qiao", "ridge")
        self.assertEqual(self.lab.read(QIAO, "ridge"), [])
        with self.assertRaises(Denied): self.lab.insert(QIAO, "N1", "ridge", "New")
    def test_09_permissive_or_leaks(self):
        self.assertEqual(ids(bad_permissive_or(self.lab, LIN)), ["H1", "H2", "R1", "R2"])
        self.assertEqual(ids(self.lab.read(LIN)), ["H1", "H2"])
    def test_10_admin_path_bypasses_subject(self):
        self.assertEqual(ids(bad_admin_path(self.lab)), ["H1", "H2", "R1", "R2"])
        self.assertEqual(ids(self.lab.read(MEI)), ["R1", "R2"])
    def test_11_shared_cache_leaks(self):
        bad_cache(self.lab, LIN)
        self.assertEqual(ids(bad_cache(self.lab, MEI)), ["H1", "H2"])
        self.assertEqual(ids(self.lab.cached_read(MEI)), ["R1", "R2"])
    def test_12_cache_revocation_invalidated(self):
        self.lab.cached_read(QIAO, "ridge")
        self.lab.revoke("qiao", "ridge")
        self.assertEqual(self.lab.cached_read(QIAO, "ridge"), [])
    def test_13_minimal_fields(self):
        self.assertIn("internal_note", bad_admin_path(self.lab)[0])
        for row in self.lab.read(LIN): self.assertEqual(set(row), set(EXPECTED["public_fields"]))
    def test_14_tenant_is_bound_parameter(self):
        self.assertEqual(self.lab.read(LIN, "ridge' OR 1=1 --"), [])
    def test_15_cache_cannot_skip_identity_gate(self):
        self.lab.cached_read(LIN)
        with self.assertRaises(Denied): self.lab.cached_read(Principal("lin", None))
    def test_16_unknown_action_denied(self):
        with self.assertRaises(Denied): self.lab.gate(EDIT, "delete")
    def test_17_shared_member_transfer_is_explicit(self):
        self.lab.update(QIAO, "H1", "ridge", "Routed")
        self.assertEqual(ids(self.lab.read(LIN)), ["H2"])
        self.assertEqual(ids(self.lab.read(MEI)), ["H1", "R1", "R2"])
    def test_18_denied_old_row_hides_existence(self):
        for ticket in ("R1", "missing"):
            with self.assertRaisesRegex(Denied, "old row denied"):
                self.lab.update(EDIT, ticket, "harbor", "Changed")

class LabContext:
    def __enter__(self):
        self.lab = Lab()
        return self.lab
    def __exit__(self, *args):
        self.lab.db.close()

if __name__ == "__main__":
    unittest.main(verbosity=2)

本次环境没有 psql 或 postgres，未安装数据库、未创建 Supabase 项目，也未执行真实 OAuth。以下是复验方案，不是已经跑通的部署脚本。具体表结构、角色、成员表权限和客户端标识需要在隔离测试项目中设计与审查。

1. 把四条工单、三位用户及四条成员关系导入测试库，先由管理员完成建表和夹具准备，再切换到实际受限角色运行请求。记录数据库版本、current_user、表所有者和角色属性。超级用户、带 BYPASSRLS 的角色以及通常可绕过策略的表所有者，不能代表普通用户验收。
2. 先启用表的 RLS，再分别配置表的 SQL 权限和 RLS 策略。只创建策略不会自动启用 RLS。SELECT 的行条件对应本例 read；INSERT 的 WITH CHECK 对应新行；UPDATE 需要 USING 检查旧行、WITH CHECK 检查新行，并核对所需 SELECT 策略。不要把 Python 的先查后写直接翻译成两个不受保护的数据库请求。
3. 设计成员表的可见范围。策略查询成员关系时，要避免意外递归、权限缺失或可被用户自己添加成员的入口；如采用辅助函数，审查其所有者、执行权限和搜索路径。本文没有提供未经执行的高权限函数模板。
4. 逐项运行 expected.json 的访问矩阵，记录查询结果、错误与写后状态；加入缺失和未知客户端、额外宽松策略、旧行允许但新行拒绝等场景。对原本拒绝的读取，数据库可能过滤成零行；非法新行常表现为策略错误，不能机械要求所有动作返回同一个异常。
5. 通过正式数据入口重复测试。管理员会话里手动设置 JWT 声明只是在模拟数据库上下文，不能证明真实令牌通过验证，更不能证明连接池不会复用上一请求的身份。至少交替执行林、梅、林的请求，并检查每次结果。
6. 同时检查直接 API、视图、函数、后台作业和缓存。对 PostgreSQL 的视图安全语义、函数执行身份和策略组合逐项核对，不让工具路径之外成为绕路入口。

PostgreSQL 行安全文档用于核对策略与角色语义；Supabase RLS 文档补充应用接入与元数据风险。用户可自行编辑的 metadata 不适合作为授权依据；即使使用受控声明，也应评估令牌更新前的陈旧窗口。

最后一轮才连接实际 MCP 客户端：先用林登录并同意授权，再用梅重复；检查拒绝同意、错误签发者、错误受众、过期令牌、错误签名，以及客户端撤销后的后续请求。记录认证失败发生在哪一层，确认未进入工具和数据库。不要将本文构造的 Principal 当成这些认证用例的替代品。

再从公网地址检查授权发现文档、401 的挑战头与反向代理后的资源地址；确认匿名发现可用，但匿名工具调用失败。客户端能完成登录不等于资源受众验证正确，工具能返回数据也不等于返回的是当前用户的数据。

Supabase 的 openid、email、profile、phone scopes 面向身份资料，不会自动给数据库表定义业务权限。client_id 可以参与行策略，却仍要与用户及租户约束共同使用。详见 Token Security & RLS。

交付验收应分成三栏保存：离线谓词模型的测试输出；真实 PostgreSQL 的角色、策略和结果；真实 OAuth/MCP 请求的认证与数据结果。本文只完成第一栏。后两栏有可执行计划，没有执行证据。在补齐前，可以用它讨论和检查授权设计，不能据此宣称 Supabase 用户隔离已上线通过。

前面的十八个测试把可信主体交给数据库，足以检查用户、客户端与租户的组合，却没有回答入口收到一种授权材料时该不该进入工具。这次沿用林与 H1，只在原 Lab 外面加一层离线准入模型：同一个人拿着面向另一个资源的材料，不能因为域名相同就被接纳；缺少操作 scope 与没有租户成员资格，也不能都用“重新登录”来恢复。

补充核对日期为 2026-10-14。官方版本页此时列出的当前协议为 2026-07-28，前文 2026-10-07 的组件核对记录保留原样。2025-06-18 授权规范已经要求客户端在授权请求和令牌请求里指明 resource、服务端验证目标受众。2025-11-25写出了运行时 scope 挑战与有限重试；2026-07-28进一步把补授权的 scope 集合写成“此前请求集合与当前挑战集合的并集”。不要把后一个版本的表述倒填成旧版本已经做过的测试。

下面没有访问服务器，也没有取得或检查任何令牌。read_grant 等名称是本地字典的标签，FixtureGrant 的每个字段都由测试作者填写。status_hint 只是适配器将来应考虑的 HTTP 状态提示，不是收到的 HTTP 响应。实验没有签名、签发者、有效期、发现文档、PKCE 或同意界面，不能用来证明 OAuth 校验、RFC 符合性或线上授权安全。

先规定两个不同的资源：工单入口 https://tickets.example.invalid/mcp 与导出入口 https://tickets.example.invalid/export。二者共享主机，但本例的服务配置明确把它们登记为不同身份。错误示范只比较 hostname，于是 other_resource 会通过，林能读到 H1、H2；修复后的模型在调用 Lab 之前拒绝它，tool_calls 为零。

这里直接比较两个预先约定的资源字符串，不是通用 URL 规范化器，也不是“JWT 的 aud 必须等于请求 URL”的规则。RFC 8707 第 2 节说明，同域有多个应用时资源标识可以包含路径，授权服务器也可以把 resource 映射为其他受众标识。真实接入应核对该部署认可的资源到受众映射，不能自行删掉路径、端口或其他区分信息。

本例自行定义 tickets:read 与 tickets:write，二者没有隐含关系，write_grant 明确持有两项；它们不是 Supabase 的身份资料 scopes。下面九行预期手工书写，既比较出口，也比较工具进入次数与 H1 写后状态。前三种失败的恢复不同：材料不适用先排查认证入口；scope 不足才可以准备补授权；客户端动作或租户关系不允许，则回到业务权限，不继续索取 scope。

admission_expected.json

[
  {"name":"read_harbor","label":"read_grant","operation":"read","tenant":"harbor","expected":{"outcome":"ok","ids":["H1","H2"]},"tool_calls":1,"h1_tenant":"harbor","h1_title":"Export stalls"},
  {"name":"same_host_wrong_resource","label":"other_resource","operation":"read","tenant":"harbor","expected":{"outcome":"authentication_denied","status_hint":401},"tool_calls":0,"h1_tenant":"harbor","h1_title":"Export stalls"},
  {"name":"no_grant","label":null,"operation":"read","tenant":"harbor","expected":{"outcome":"authentication_denied","status_hint":401},"tool_calls":0,"h1_tenant":"harbor","h1_title":"Export stalls"},
  {"name":"unknown_grant","label":"unknown","operation":"read","tenant":"harbor","expected":{"outcome":"authentication_denied","status_hint":401},"tool_calls":0,"h1_tenant":"harbor","h1_title":"Export stalls"},
  {"name":"read_cannot_update","label":"read_grant","operation":"update","tenant":"harbor","expected":{"outcome":"insufficient_scope","status_hint":403,"scope":["tickets:write"]},"tool_calls":0,"h1_tenant":"harbor","h1_title":"Export stalls"},
  {"name":"manual_recovery","label":"write_grant","operation":"update","tenant":"harbor","expected":{"outcome":"ok","updated":"H1"},"tool_calls":1,"h1_tenant":"harbor","h1_title":"Reviewed"},
  {"name":"scopes_do_not_add_membership","label":"write_grant","operation":"update","tenant":"ridge","expected":{"outcome":"business_denied"},"tool_calls":1,"h1_tenant":"harbor","h1_title":"Export stalls"},
  {"name":"scopes_do_not_upgrade_client","label":"report_with_write","operation":"update","tenant":"harbor","expected":{"outcome":"business_denied"},"tool_calls":1,"h1_tenant":"harbor","h1_title":"Export stalls"},
  {"name":"unknown_operation","label":"write_grant","operation":"delete","tenant":"harbor","expected":{"outcome":"unsupported_operation"},"tool_calls":0,"h1_tenant":"harbor","h1_title":"Export stalls"}
]

把下面两个 Python 文件和 admission_expected.json 放在前文三个文件旁边。Admission 只接受夹具标签；经过资源和 scope 两项模型检查，才把同一个 Principal 送进原 Lab。它没有把 scope 换算成 editor，也没有替林添加山脊成员。进入工具后抛出的 Denied 被记为 business_denied，本文不替它选择真实 MCP 错误编码。

RFC 6750 的错误规则区分无效材料的 401 与权限范围不足的 403。真实适配器还要正确构造和解析 WWW-Authenticate，而本实验只交换已结构化的字典，不声称验证了挑战头。ManualStepUp 只在本地案例明确传入布尔值 True 时返回人工提案；它没有打开登录页、同意权限、获取凭据或重发请求的能力。False 在本夹具中表示拒绝，None 表示取消，其他类型都无效；它们都会关闭当前对象中该资源与操作的尝试。尚未作出决定时不要调用 plan，不能把 None 当作“继续等待”。

admission_contract.py

"""Synthetic admission contract only. No token parsing, OAuth, HTTP or crypto."""
from dataclasses import dataclass
from authorization_lab import Lab, Principal, Denied, ids

RESOURCE = "https://tickets.example.invalid/mcp"
OTHER_RESOURCE = "https://tickets.example.invalid/export"
READ = "tickets:read"
WRITE = "tickets:write"
NEEDED = {"read": frozenset({READ}), "update": frozenset({WRITE})}

@dataclass(frozen=True)
class FixtureGrant:
    # All fields are authored by the test. This type is NOT a validator output.
    resource: str
    scopes: frozenset[str]
    principal: Principal

EDITOR = Principal("lin", "editor")
GRANTS = {
    "read_grant": FixtureGrant(RESOURCE, frozenset({READ}), EDITOR),
    "write_grant": FixtureGrant(RESOURCE, frozenset({READ, WRITE}), EDITOR),
    "other_resource": FixtureGrant(OTHER_RESOURCE, frozenset({READ}), EDITOR),
    "report_with_write": FixtureGrant(RESOURCE, frozenset({READ, WRITE}),
                                       Principal("lin", "report")),
}

class Admission:
    def __init__(self):
        self.lab = Lab()
        self.tool_calls = 0

    def call(self, label, operation, tenant="harbor"):
        # Inputs are fixture labels, never bearer credentials or decoded JWTs.
        grant = GRANTS.get(label)
        if grant is None or grant.resource != RESOURCE:
            return {"outcome": "authentication_denied", "status_hint": 401}
        if operation not in NEEDED:
            return {"outcome": "unsupported_operation"}
        needed = NEEDED[operation]
        if not needed <= grant.scopes:
            return {"outcome": "insufficient_scope", "status_hint": 403,
                    "scope": sorted(needed)}
        self.tool_calls += 1
        try:
            if operation == "read":
                return {"outcome": "ok", "ids": ids(self.lab.read(grant.principal, tenant))}
            self.lab.update(grant.principal, "H1", tenant, "Reviewed")
            return {"outcome": "ok", "updated": "H1"}
        except Denied:
            # Domain denial, not an OAuth scope challenge. No transport encoding here.
            return {"outcome": "business_denied"}

    def close(self):
        self.lab.db.close()

class ManualStepUp:
    """One local attempt; one terminal decision per resource/operation. No I/O."""
    def __init__(self):
        self.decided = set()

    def plan(self, result, operation, previously_requested, *, approved):
        key = (RESOURCE, operation)
        if operation not in NEEDED or key in self.decided:
            return {"outcome": "stop"}
        # A declined/cancelled/invalid decision also closes this local attempt.
        self.decided.add(key)
        if (result.get("outcome") != "insufficient_scope"
                or approved is not True):
            return {"outcome": "stop"}
        scopes = set(previously_requested) | set(result["scope"])
        return {"outcome": "manual_proposal", "resource": RESOURCE,
                "operation": operation, "scope": sorted(scopes)}

test_admission_contract.py

"""Offline tests; labels and approvals are synthetic fixtures, not authorization."""
import json
from pathlib import Path
import unittest
from urllib.parse import urlsplit
from admission_contract import Admission, GRANTS, RESOURCE, READ, WRITE, ManualStepUp

CASES = json.loads(Path(__file__).with_name("admission_expected.json").read_text())

class AdmissionTest(unittest.TestCase):
    def test_manual_matrix(self):
        for case in CASES:
            with self.subTest(case=case["name"]):
                a = Admission()
                try:
                    got = a.call(case["label"], case["operation"], case["tenant"])
                    self.assertEqual(got, case["expected"])
                    self.assertEqual(a.tool_calls, case["tool_calls"])
                    row = a.lab.raw("H1")
                    self.assertEqual(row["tenant_id"], case["h1_tenant"])
                    self.assertEqual(row["title"], case["h1_title"])
                finally:
                    a.close()

    def test_same_host_is_not_resource_identity(self):
        # Intentionally broken test-only admission: compare just the host.
        grant = GRANTS["other_resource"]
        bad_admitted = urlsplit(grant.resource).hostname == urlsplit(RESOURCE).hostname
        self.assertTrue(bad_admitted)
        a = Admission()
        try:
            if bad_admitted:
                leaked = a.lab.read(grant.principal, "harbor")
            self.assertEqual([row["id"] for row in leaked], ["H1", "H2"])
            self.assertEqual(a.call("other_resource", "read")["outcome"],
                             "authentication_denied")
            self.assertEqual(a.tool_calls, 0)
        finally:
            a.close()

    def test_explicit_recovery_preserves_read_scope(self):
        a, flow = Admission(), ManualStepUp()
        try:
            first = a.call("read_grant", "update")
            self.assertEqual(a.tool_calls, 0)
            self.assertEqual(flow.plan(first, "update", [READ], approved=True), {
                "outcome": "manual_proposal", "resource": RESOURCE,
                "operation": "update", "scope": [READ, WRITE]})
            # Test explicitly selects a different fixture, no token obtained here.
            self.assertEqual(a.call("write_grant", "update"),
                             {"outcome": "ok", "updated": "H1"})
            self.assertEqual(a.tool_calls, 1)
            self.assertEqual(a.lab.raw("H1")["title"], "Reviewed")
        finally:
            a.close()

    def test_decline_stops_without_a_tool_call(self):
        a, flow = Admission(), ManualStepUp()
        try:
            first = a.call("read_grant", "update")
            self.assertEqual(flow.plan(first, "update", [READ], approved=False),
                             {"outcome": "stop"})
            # A later callback cannot reopen this same declined attempt.
            self.assertEqual(flow.plan(first, "update", [READ], approved=True),
                             {"outcome": "stop"})
            self.assertEqual(a.tool_calls, 0)
            self.assertEqual(a.lab.raw("H1")["title"], "Export stalls")
        finally:
            a.close()

    def test_non_boolean_approval_cannot_propose(self):
        a = Admission()
        try:
            first = a.call("read_grant", "update")
            for value in ("false", "true", 1, 1.0, [True], {"approved": True},
                          0, "", [], {}):
                with self.subTest(value=repr(value)):
                    flow = ManualStepUp()
                    self.assertEqual(flow.plan(first, "update", [READ], approved=value),
                                     {"outcome": "stop"})
                    self.assertEqual(flow.plan(first, "update", [READ], approved=True),
                                     {"outcome": "stop"})
            self.assertEqual(a.tool_calls, 0)
            self.assertEqual(a.lab.raw("H1")["title"], "Export stalls")
        finally:
            a.close()

    def test_cancellation_is_terminal_for_attempt(self):
        a, flow = Admission(), ManualStepUp()
        try:
            first = a.call("read_grant", "update")
            # None is this fixture's cancellation marker, not a pending decision.
            self.assertEqual(flow.plan(first, "update", [READ], approved=None),
                             {"outcome": "stop"})
            self.assertEqual(flow.plan(first, "update", [READ], approved=True),
                             {"outcome": "stop"})
            self.assertEqual(a.tool_calls, 0)
            self.assertEqual(a.lab.raw("H1")["title"], "Export stalls")
        finally:
            a.close()

    def test_second_scope_failure_stops(self):
        a, flow = Admission(), ManualStepUp()
        try:
            first = a.call("read_grant", "update")
            flow.plan(first, "update", [READ], approved=True)
            again = a.call("read_grant", "update")
            self.assertEqual(flow.plan(again, "update", [READ, WRITE], approved=True),
                             {"outcome": "stop"})
            self.assertEqual(a.tool_calls, 0)
        finally:
            a.close()

    def test_business_denial_does_not_request_more_scopes(self):
        a, flow = Admission(), ManualStepUp()
        try:
            result = a.call("write_grant", "update", "ridge")
            self.assertEqual(result, {"outcome": "business_denied"})
            self.assertEqual(flow.plan(result, "update", [READ, WRITE], approved=True),
                             {"outcome": "stop"})
            self.assertEqual(a.lab.raw("H1")["tenant_id"], "harbor")
        finally:
            a.close()

    def test_authentication_failure_does_not_step_up(self):
        a, flow = Admission(), ManualStepUp()
        try:
            result = a.call("other_resource", "read")
            self.assertEqual(flow.plan(result, "read", [READ], approved=True),
                             {"outcome": "stop"})
            self.assertEqual(a.tool_calls, 0)
        finally:
            a.close()

if __name__ == "__main__":
    unittest.main(verbosity=2)

运行 python -m unittest -v test_authorization test_admission_contract。预期是 Ran 27 tests 和 OK：原十八个方法不改动，入口补充现在共九个方法，其中矩阵方法包含九个独立案例。测试故意让 hostname 错误检查放行，再确认准入模型拒绝；因此绿灯包含“成功重现错误”，并不是说坏检查可以用于生产。

1. 先用 read_grant 更新海港 H1：得到 insufficient_scope、status_hint=403、scope=[tickets:write]，tool_calls=0，标题仍为 Export stalls。
2. 合成案例明确批准一次提案后，ManualStepUp 返回 resource、update 及 tickets:read、tickets:write 两项。测试随后手工选择 write_grant 再调用，H1 标题变为 Reviewed，累计只进入工具一次。选择新夹具不代表已经发生真实授权。
3. 把 approved 改成 False，或用 None 表示取消，当前尝试立即结束；随后在同一个对象上改传 True，也仍然停止。非布尔值同样不能生成提案。即使第一次批准过，若仍选 read_grant，第二次 scope 失败也停止。每个资源与操作只作一次终态决定是本例的保守上限，不是协议规定的固定次数。
4. 拿 write_grant 把 H1 移到 ridge，或给 report 客户端相同两项 scope，都会进入业务检查后拒绝，H1 原样保留。ManualStepUp 对这两类业务拒绝都不提出补授权。

2026-10-20 回归修正：旧版用 if not approved 判断，字符串 "false"、数字 1 等真值对象也会生成提案；拒绝后没有记录终态，同一个对象稍后又传 True 也能生成提案。这两种情况没有执行工具，但违背了本例“明确批准且拒绝即结束”的约定。Python 的真值规则不等于业务批准，因此现在使用 approved is True，并在检查决定前记录该尝试已经结束。新增回归覆盖十个非布尔输入、取消后迟到的 True，以及拒绝后迟到的 True；原来的明确批准与第二次失败测试仍保留。

这个终态只存在于当前 ManualStepUp 对象内，不是持久化同意记录，也不处理并发回调。调用方必须保留该次尝试的对象；不能自动新建对象来绕过停止。若用户另行发起一次新尝试，应由真实产品流程重新取得适用于该次尝试的决定。这里的布尔值始终只是作者手写的测试夹具，不是用户授权证据。

这也解释了为什么不能把 403 一律接到自动登录。应先识别可信响应的具体错误，再由用户决定是否增加授权；拒绝或取消便结束当前尝试，不能改走管理员连接。对真实写请求还要分别解决重放和幂等问题，本例只在明确尚未进入工具的 scope 失败后演示人工恢复，不给网络超时后的重试作保证。

2026-10-14 的记录保留为当时对源文件、HTML 提取文件和干净解压包的验证。2026-10-20 对修正后的本页文件重新运行普通模式与 python -O：二十七个测试方法通过；旧实现配新测试会失败，恢复真值判断、推迟终态记录、移除重复调用拦截三种单点变异也均被捕获。这次只验证本页教学代码与内存 SQLite，不把旧下载包的记录当成修正版的验证。上线前仍须补齐前文三栏证据中的真实 OAuth 与 PostgreSQL 两栏，包括资源到受众映射、签名与签发者验证、拒绝同意、代理后的挑战头、受限数据库角色及写后状态。当前版本还有 scope 层级等要求；这个只有两项互不包含权限的模型没有验证层级实现，也没有验证多实例撤销、并发或浏览器兼容性。