AgentDepthAgent 工程学习指南
输入概念或问题,检索正文和待编写目录。
第一次学习?从 L0 总览开始 →

工具权限:先确认身份,再检查资源访问

我的学习 →

正在读取学习记录…

1. 一个工具调用需要哪些权限

权限描述“谁可以对哪个对象做什么”。例如Alice拥有一篇笔记,她可以读取正文;Bob虽然也能使用笔记查询工具,却需要相应笔记的访问权限。工具名称、用户身份和具体资源共同决定这次请求是否放行。

身份认证确认来访者是谁,授权判断来访者可做什么。把它们想成图书馆里的借书证和借阅范围:办好证之后,每次借阅还要看这本书是否在允许范围内。

可信身份进入权限检查,目标笔记归属和租户参与比较,通过后才读取正文
先读归属元数据,再检查访问条件,最后加载正文。拒绝分支在正文读取之前结束。

租户是应用划分的组织或团队空间。一个用户可能参与多个团队,笔记属于某个空间。示例要求身份有效、具有 note:read 操作范围,同时用户和租户都与笔记一致。scope(操作范围)说明可使用哪类操作,资源归属进一步说明可读取哪篇笔记。

管理员核对来访者与资料柜的归属后决定是否打开柜子
能使用查询服务之后,仍要核对当前对象的归属。

2. 运行五个访问请求

在同一新文件夹建立 permissions_lab.py 和下面的测试文件。程序内置Alice和Bob的笔记,正文读取函数会计数,方便看见拒绝发生的位置。

代码里的 Principal 保存当前登录者:user_id 是用户编号,tenant 是所属组织,active 表示账号是否可用,scopes 保存允许的操作,例如 note:read 表示读取笔记。笔记编号则放在工具参数里,用来指明这次读哪一篇。

@dataclass 自动生成初始化等常用方法,所以可以用 Principal(...) 创建身份记录。frozen=True 让实例创建后保持字段值稳定,普通赋值会报错。frozenset 是元素不重复、创建后保持不变的集合,适合保存这份身份的操作范围。是否允许访问,随后由 permits 根据身份和笔记归属判断。

先在 PowerShell 创建本篇的独立目录并进入;路径可换成你自己的非系统盘位置:

Bash
mkdir -Force "F:\学习\agentdepth-l2-permissions"
cd "F:\学习\agentdepth-l2-permissions"

将本篇的 permissions_lab.py、test_permissions_lab.py 保存到这个目录。保存全部文件后运行 dir *.py,核对上述文件名,再执行下文命令;新开窗口时先回到该目录,并按 Python 环境篇核对 python 的解释器路径。

Python
"""根据身份、租户和资源归属,在读取前检查权限。"""
from dataclasses import dataclass
import json


@dataclass(frozen=True)
class Principal:
    user_id: str
    tenant: str
    active: bool
    scopes: frozenset[str]


# 资源归属从服务端元数据读取,与当前身份进行比较。
METADATA = {'a-1': ('alice', 'team-a'), 'a-2': ('bob', 'team-a'), 'b-1': ('alice', 'team-b')}
BODIES = {'a-1': 'Alice fixture note', 'a-2': 'Bob fixture note', 'b-1': 'Other tenant fixture'}


# 先检查身份状态和操作范围,再比较拥有者与租户。
def permits(principal, metadata):
    return (principal.active and 'note:read' in principal.scopes
            and metadata == (principal.user_id, principal.tenant))


# 校验参数与权限通过后,才调用正文读取函数。
def execute(principal, tool, args, load_body=BODIES.__getitem__, policy=permits):
    if tool != 'read_note':
        return {'status': 'unknown_tool'}
    if (not isinstance(args, dict) or set(args) != {'note_id'}
            or not isinstance(args['note_id'], str) or not 1 <= len(args['note_id']) <= 80):
        return {'status': 'invalid_arguments'}
    metadata = METADATA.get(args['note_id'])
    try:
        allowed = metadata is not None and policy(principal, metadata)
    except RuntimeError:
        return {'status': 'authorization_unavailable'}
    if not allowed:
        return {'status': 'denied'}
    return {'status': 'ok', 'body': load_body(args['note_id'])}


def main():
    alice = Principal('alice', 'team-a', True, frozenset({'note:read'}))
    for label, tool, args in [
        ('own', 'read_note', {'note_id': 'a-1'}),
        ('same-tenant-other-owner', 'read_note', {'note_id': 'a-2'}),
        ('other-tenant', 'read_note', {'note_id': 'b-1'}),
        ('spoofed-user', 'read_note', {'note_id': 'a-2', 'user_id': 'bob'}),
        ('unknown-tool', 'delete_note', {'note_id': 'a-1'}),
    ]:
        reads = []
        result = execute(alice, tool, args, load_body=lambda key: reads.append(key) or BODIES[key])
        print(json.dumps({'case': label, 'result': result, 'body_reads': len(reads)}, ensure_ascii=False))


if __name__ == '__main__':
    main()
Bash
python permissions_lab.py

输出:

TXT
{"case": "own", "result": {"status": "ok", "body": "Alice fixture note"}, "body_reads": 1}
{"case": "same-tenant-other-owner", "result": {"status": "denied"}, "body_reads": 0}
{"case": "other-tenant", "result": {"status": "denied"}, "body_reads": 0}
{"case": "spoofed-user", "result": {"status": "invalid_arguments"}, "body_reads": 0}
{"case": "unknown-tool", "result": {"status": "unknown_tool"}, "body_reads": 0}

第一行 own 读取Alice自己的笔记,body_reads 为1。后四行分别是同团队的他人笔记、其他团队笔记、参数夹带用户字段、未知工具;正文读取次数都为0。

Principal 保存可信身份。METADATA 保存笔记归属,BODIES 保存正文。permits 比较身份与元数据。execute 先校验工具和参数,再读取归属并授权,通过后才调用 load_body。把身份对象放在参数之外,便于服务端从已验证会话取得它。

args 只接受 note_id,因此尝试加 user_id 会产生 invalid_arguments。正文读取函数可注入,这使测试能检查“被拒绝时是否读过正文”,而不仅仅检查返回字符串。

3. 修改身份,观察访问结果

找到 main 中Alice的身份,把 scopes 的 note:read 去掉,再运行。原来自己的笔记也会被拒绝,读取次数为0。恢复范围,把身份的租户改为另一个团队,再观察归属比较的变化。

新增允许共享的规则时,先定义共享记录属于哪一个租户,再让权限判断比较共享者。保留已有测试,确认共享只改变目标请求的结果。

4. 运行检查用例

把下方测试文件保存到程序所在目录:

下面的测试检查自己的笔记、他人的笔记、账号状态和操作权限,并核对拒绝后正文读取次数。

测试代码:test_permissions_lab.py
Python
# 核对MCP认证与授权的正常结果和错误分支;断言失败时会显示对应的测试。
import unittest
from dataclasses import replace
from permissions_lab import Principal, execute


class PermissionTests(unittest.TestCase):
    def setUp(self):
        self.user = Principal('alice', 'team-a', True, frozenset({'note:read'}))
        self.reads = []

    def run_request(self, **changes):
        values = dict(principal=self.user, tool='read_note', args={'note_id': 'a-1'},
                      load_body=lambda note_id: self.reads.append(note_id) or 'safe fixture')
        values.update(changes)
        return execute(**values)

    def test_own_resource(self):
        self.assertEqual(self.run_request()['status'], 'ok')
        self.assertEqual(self.reads, ['a-1'])

    def test_same_tenant_other_owner(self):
        self.assertEqual(self.run_request(args={'note_id': 'a-2'})['status'], 'denied')
        self.assertEqual(self.reads, [])

    def test_other_tenant(self):
        self.assertEqual(self.run_request(args={'note_id': 'b-1'})['status'], 'denied')
        self.assertEqual(self.reads, [])

    def test_model_cannot_supply_identity(self):
        self.assertEqual(self.run_request(args={'note_id': 'a-2', 'user_id': 'bob'})['status'], 'invalid_arguments')
        self.assertEqual(self.reads, [])

    def test_unknown_tool(self):
        self.assertEqual(self.run_request(tool='delete_note')['status'], 'unknown_tool')
        self.assertEqual(self.reads, [])

    def test_missing_scope_or_revocation(self):
        for user in [replace(self.user, scopes=frozenset()), replace(self.user, active=False)]:
            self.assertEqual(self.run_request(principal=user)['status'], 'denied')
        self.assertEqual(self.reads, [])

    def test_missing_resource_same_public_denial(self):
        self.assertEqual(self.run_request(args={'note_id': 'missing'}), self.run_request(args={'note_id': 'a-2'}))

    def test_policy_error_fails_closed(self):
        def unavailable(*_):
            raise RuntimeError('private backend details')
        result = self.run_request(policy=unavailable)
        self.assertEqual(result, {'status': 'authorization_unavailable'})
        self.assertEqual(self.reads, [])


if __name__ == '__main__':
    unittest.main()
Bash
python -m unittest test_permissions_lab -v

测试同时检查正常和失败路径。先恢复刚才的改动,再运行;若出现导入错误,核对文件位置与运行目录。

工具请求与服务端身份分别进入检查,核对归属后允许分支加载正文
请求带目标,服务端带身份。授权通过后加载正文。

5. 把检查放到每个执行入口

入口取得身份的位置授权时机
Web请求已验证的会话或访问令牌执行工具之前
后台队列任务绑定身份,再核对当前权限出队执行之前
批量处理每一项资源的归属记录逐项读取或写入之前
人工批准写入审批者与操作提案批准后、实际执行前

权限服务异常时,程序返回 authorization_unavailable,保留等待核对或重新检查的分支。示例只转换特定的 RuntimeError,其他代码错误适合由外层日志定位。资源不存在和无权访问对外都可采用统一拒绝提示,内部保留必要分类。

真实数据库可将租户和归属直接放进读取条件,减少检查与加载之间的数据变化窗口。排队任务执行前重新核对权限,也能处理用户刚被移出团队的情况。OWASP 授权指南进一步介绍默认拒绝和逐请求检查。

总结:身份来自验证后的服务端上下文,参数标明目标,归属和范围决定访问,正文在通过后读取。继续读 并行工具。

课后练习展开习题收起习题

4 道练习 · 客观题规则评分 · 其余题目待批改

正在读取本地记录…

不调用模型,不执行代码。简答、代码和场景题仅保存答案并展示自查标准,不生成评分、徽章或认证。输入内容不会自动保存,请点击提交;不要填写个人信息或密钥。

练习 1 / 4 · v1.0.0

选择题 · 白名单为什么没挡住越权?

read_note在白名单中,note_id类型正确。Alice想读取Bob的笔记,还必须检查什么?

对应知识点:L2-permissions

选择一个答案
练习 2 / 4 · v1.0.0

简答题 · 谁提供当前用户身份?

模型参数带user_id=bob,但服务端会话属于Alice。为什么不能采用模型字段?

对应知识点:L2-permissions

练习 3 / 4 · v1.0.0

代码题 · 证明拒绝后没有读正文

修改离线实验,为同租户不同所有者和跨租户各写测试,除了denied还应断言什么?本站不运行代码。

对应知识点:L2-permissions

练习 4 / 4 · v1.0.0

场景题 · 排队后权限被撤销

任务入队时用户有权限,十分钟后被撤销。worker能直接使用入队时的允许结论吗?授权服务又恰好不可用怎么办?

对应知识点:L2-permissions

图片查看器
100%

文章配图 · 按 Esc 关闭