MCP认证与授权:确认来访者和资源范围
正在读取学习记录…
1. 先确认谁在调用
远端服务需要知道来访者身份,再判断本次调用能访问什么。认证确认身份,授权判断操作和资源。访问令牌是客户端提交给服务端的一种凭证,服务端验证后取得相应身份与范围。
可以把令牌理解成有适用地点和有效期的通行证。服务端核对签发方、接收方、有效期和操作范围,再按当前资源判断允许的动作。用户在聊天里写“我是管理员”,只是一段输入文字。
MCP2025-11-25的授权规范针对HTTP传输定义OAuth相关流程。OAuth可以让应用取得有范围的访问权限。stdio通常通过启动环境等方式取得凭据,配置方式与HTTP令牌流程分别处理。
2. 先运行资源授权判断
下面用已取得的Principal身份对象,学习认证之后的资源判断。它包含用户、租户、是否有效和操作范围。新建permissions_lab.py:
先在 PowerShell 创建本篇的独立目录并进入;路径可换成你自己的非系统盘位置:
mkdir -Force "F:\学习\agentdepth-l2-mcp-auth"
cd "F:\学习\agentdepth-l2-mcp-auth"将本篇的 permissions_lab.py、test_permissions_lab.py 保存到这个目录。保存全部文件后运行 dir *.py,核对上述文件名,再执行下文命令;新开窗口时先回到该目录,并按 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()python permissions_lab.py输出:
{"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成功,跨拥有者和跨租户请求被拒绝;spoofed-user夹带身份字段,参数校验拒绝。body_reads检查真正的正文加载次数。
Principal代表认证完成后的主体,METADATA代表资源归属。permits比较身份、范围和归属。HTTP服务可以把验证后的令牌主体转换成类似对象,再交给工具执行入口。
把active改为False,自己的笔记也会拒绝;恢复后移除note:read,同样拒绝。认证身份有效和操作范围足够,是分别检查的条件。
4. 把授权放入HTTP请求流程
| 步骤 | 读取什么 | 得到什么 |
|---|---|---|
| 验证令牌 | 签发方、接收方、有效期 | 可信身份 |
| 核对操作范围 | scope与工具动作 | 允许的操作类别 |
| 核对资源 | 租户、归属、共享关系 | 这次对象是否可访问 |
| 执行工具 | 已校验参数与身份 | 业务结果 |
| 记录结果 | 任务、调用和必要错误码 | 可回查记录 |
audience或资源标识表达令牌的适用服务,服务端需要验证它。令牌应发给其目标服务;服务调用下游时应使用下游认可的凭证和范围。到期、撤销和权限变化都需要在请求路径中处理。
令牌与密钥存储在受控位置,日志用脱敏编号,错误消息保留必要分类。权限服务暂时异常时,等待恢复或人工核对当前请求;后台队列在真正执行时重新核验权限。
总结:认证确认来访者,范围限定操作,归属限定对象,执行入口逐请求核对。回到 L2总览,再把工具、权限和协议串起来。
5. 验证消息与边界
测试文件保存到同一目录:
下面的测试检查主体是否启用、是否拥有读取范围,以及笔记的租户和归属;还检查正文只在授权通过后加载。令牌的签名与目标服务检查属于前面的HTTP认证流程。
测试代码:test_permissions_lab.py
# 核对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()python -m unittest test_permissions_lab -v恢复实验中的改动后运行测试。失败时先查看对应参数、编号或权限条件,再核对程序分支。
课后练习展开习题收起习题
正在读取本地记录…
不调用模型,不执行代码。简答、代码和场景题仅保存答案并展示自查标准,不生成评分、徽章或认证。输入内容不会自动保存,请点击提交;不要填写个人信息或密钥。
选择题 · 先判断边界
关于“MCP 授权与信任边界”,哪种做法更稳妥?
对应知识点:L2-mcp-auth
简答题 · 解释失败
写出一个失败输入并说明记录字段。
对应知识点:L2-mcp-auth
代码题 · 边界测试
补充正常、边界、失败三组输入和预期结果。
对应知识点:L2-mcp-auth
场景题 · 结果未知
证据不足时如何设计状态和人工接管?
对应知识点:L2-mcp-auth
描述会发送给本站维护者,用于修订文章,保存90天。请只填写文章问题,省略个人信息、密钥和内部资料。