FastAPI的行级权限
fastapi_permissions的Python项目详细描述
fastapi的行级权限
在尝试优秀的fastapi框架时,我缺少了一种平静:一种简单、声明性的方式来定义用户(和角色/组)对资源的权限。因为我真的很喜欢金字塔处理这个问题的方式,所以我重新实现并调整了fastapi的系统(好吧,你可以称之为公然的剽窃)。
一个非常简单且不完整的示例:
fromfastapiimportDepends,FastAPIfromfastapi.securityimportOAuth2PasswordBearerfromfastapi_permissionsimportconfigure_permissions,Allow,DenyfrompydanticimportBaseModelapp=FastAPI()oauth2_scheme=OAuth2PasswordBearer(tokenUrl="/token")classItem(BaseModel):name:strowner:strdef__acl__(self):return[(Allow,Authenticated,"view"),(Allow,"role:admin","edit"),(Allow,f"user:{self.owner}","delete"),]classUser(BaseModel):name:strdefprincipals(self):return[f"user:{self.name}"]defget_current_user(token:str=Depends(oauth2_scheme)):...defget_active_user_principals(user:User=Depends(get_current_user)):...defget_item(item_identifier):...# Permission is already wrapped in Depends()Permission=configure_permissions(get_active_user_principals)@app.get("/item/{item_identifier}")asyncdefshow_item(item:Item=Permission("view",get_item)):return[{"item":item}]
要获得更好的示例,请在虚拟环境中安装fastapi_permissions
source(请参见下面的进一步内容),然后启动测试服务器:
(permissions) $ uvicorn fastapi_permissions.example:app --reload
访问http://127.0.0.1:8000/docs" rel="nofollow">http://127.0.0.1:8000/docs进行尝试。有两个用户:"bob"和"alice",都有密码"secret"。
这个示例是从fastapi示例派生的,所以应该很熟悉。新添加的内容在源文件 对于大多数应用程序,使用作用域来确定用户的权限就足够了。因此,如果作用域适合您的应用程序,请使用它们-它们已经是FastAPI框架的一部分。 虽然作用域仅与用户的状态相关联,但也有 让我们以一篇科学论文为例:根据提交过程的状态(如"草稿"、"已提交"、"同行评审"或"已发布"),不同的用户应该具有不同的查看、编辑或收回权限。这可以与路径定义函数中的自定义代码一起实现,但fastapi权限提供了一种在单个位置定义这些约束的方法。 第二种情况是,FastAPI权限可能是您应用程序的正确附加功能:如果您的大脑像我的大脑一样连接/预处理到这样的权限模型-例如,长时间暴露在金字塔中../p> 长话短说:在您需要其他功能之前,请使用作用域。 由于 系统依赖于fastapi中没有的几个概念: 资源通过其 示例: 您不需要在访问控制列表的末尾添加任何"deny all子句",这是自动暗示的。acl中的所有条目都按列表中提供的顺序签入这使得一些复杂的配置变得简单,但有时也会使下背部疼痛… 这两个原则 必须提供返回当前活动用户主体的函数。主体只是一个字符串列表,标识用户和用户所属的组/角色: 示例: 有两个特殊的主体也有助于提供访问控制列表: 应添加 权限只是表示要对资源执行的操作的字符串。只要编点东西就行了。 与特殊主体一样,有一个可用作通配符的特殊权限: 在使用权限系统之前,您必须提供以下内容: 具有一些默认值的简单配置: 提供一个配置选项: 若要在路径操作中使用访问控制,请使用权限和资源调用未配置的函数。如果授予了权限,则将返回选中该权限的请求资源,或者在本例中返回acl列表 与直接使用访问控制列表不同,您还可以提供一个依赖函数,该函数可以从数据库中获取资源,资源应该通过 有时您可能希望检查函数内部的权限,而不是路径操作的定义: 使用 提供的另一个功能是 请注意, 主要工作是在 前面在路径操作定义中使用的 或者换句话说:要有一个好的api,path操作函数中的 应使用虚拟环境进行测试和开发。 开发需要安装flit: 然后可以使用 如果您还可以运行所有测试并使用 当准备好作为已安装的软件包测试所有内容时(如果
使用fastapi_permissions/example.py中用注释标记
为什么不使用作用域?
fastapi_权限
请考虑所请求资源的状态。
概念
fastapi_权限
主要来自金字塔框架,如果您不清楚,我强烈建议查看它的安全文档。资源和访问控制列表
\u acl\u
属性提供访问控制列表。它可以是对象的属性,也可以是可调用的。列表中的每个条目都是一个包含三个值的元组:fastapi_permissions.allow
或fastapi_permissions.deny
fromfastapi_permissionsimportAllow,Deny,Authenticated,EveryoneclassStaticAclResource:__acl__=[(Allow,Everyone,"view"),(Allow,"role:user","share")]classDynamicAclResource:def__acl__(self):return[(Allow,Authenticated,"view"),(Allow,"role:user","share"),(Allow,f"user:{self.owner}","edit"),]# in contrast to pyramid, resources might be access conroll list themselves# this can save some typing:AclResourceAsList=[(Allow,Everyone,"view"),(Deny,"role:troll","edit")]
Everyone
和Authenticated
将在短时间内讨论。用户和主体标识符
defget_active_principals(user:User=Depends(get_current_user)):ifuser:# user is logged inprincipals=[Everyone,Authenticated]principals.extend(getattr(user,"principals",[]))else:# user is not logged inprincipals=[Everyone]returnprincipals
特别负责人
Everyone
和Authenticated
Everyone
主体,而不考虑任何其他定义的主体或登录状态,Authenticated
仅应为登录的用户添加。权限
fastapi\u permissions.all
用法
配置权限系统
fromfastapi_permissionsimportconfigure_permissions# must be provideddefget_active_principals(...):""" returns the principals of the current logged in user"""...permission=configure_permissions(get_active_principals)
fromfastapi_permissionsimportconfigure_permissions# must be provideddefget_active_principals(...):""" returns the principals of the current logged in user"""...permission=configure_permissions(get_active_principals,permission_exception)
在路径操作中使用权限
fromfastapi_permissionsimportconfigure_permissions,Allow# must be provideddefget_active_principals(...):""" returns the principals of the current logged in user"""...example_acl=[(Allow"role:user","view")]# Permission is already wrapped in Depends()Permission=configure_permissions(get_active_principals)@app.get("/")asyncdefroot(acls:list=Permission("view",example_acl)):return{"OK"}
\uu acl\
属性提供其访问控制列表:fromfastapi_permissionsimportconfigure_permissions,Allow# must be provideddefget_active_principals(...):""" returns the principals of the current logged in user"""...# fetches a resource from the databasedefget_item(item_id:int):""" returns a resource from the database The resource provides an access controll list via its "__acl__" attribute. """...# Permission is alredy wrapped in Depends()Permission=configure_permissions(get_active_principals)@app.get("/item/{item_id}")asyncdefshow_item(item:Item=permission("view",get_item)):return{"item":item}
助手函数
has_permission(用户主体、权限、资源)
可以编程方式执行权限检查。函数签名可以很容易地用"john eat apple?"。结果将是true
或false
,因此不需要try/except块。
fromfastapi_permissionsimport(has_permission,Allow,All,Everyone,Authenticated)user_principals==[Everyone,Authenticated,"role:owner","user:bob"]apple_acl==[(Allow,"role:owner",All)]ifhas_permission(user_principals,"eat",apple_acl):print"Yum!"
列出权限(用户主体、资源)
如果权限被授予或拒绝,将返回所有可用权限的dict和布尔值:fromfastapiimportDepends,FastAPIfromfastapi.securityimportOAuth2PasswordBearerfromfastapi_permissionsimportconfigure_permissions,Allow,DenyfrompydanticimportBaseModelapp=FastAPI()oauth2_scheme=OAuth2PasswordBearer(tokenUrl="/token")classItem(BaseModel):name:strowner:strdef__acl__(self):return[(Allow,Authenticated,"view"),(Allow,"role:admin","edit"),(Allow,f"user:{self.owner}","delete"),]classUser(BaseModel):name:strdefprincipals(self):return[f"user:{self.name}"]defget_current_user(token:str=Depends(oauth2_scheme)):...defget_active_user_principals(user:User=Depends(get_current_user)):...defget_item(item_identifier):...# Permission is already wrapped in Depends()Permission=configure_permissions(get_active_user_principals)@app.get("/item/{item_identifier}")asyncdefshow_item(item:Item=Permission("view",get_item)):return[{"item":item}]
0
"permissions:*"
是fastapi\u permissions.all的字符串表示形式。
工作原理
has\u permissions()
函数中完成的,但最有趣的(至少对我来说)是configure\u permissions()
和permission\u dependency\u factory()
函数。permission()
thingy实际上是所提到的permission\u dependency\u factory()
。configure_permissions()
函数只是使用functools.partial为它提供一些默认值。这将函数签名从
权限依赖工厂(权限、资源、活动的主体函数、权限异常)减少到
部分函数(权限、资源)<·代码>< /P>
权限依赖项工厂
返回另一个具有签名的函数权限依赖项(依赖(资源)、依赖(活动的原则)
。这是acutal签名,它在路径操作定义中用于搜索和注入依赖项。剩下的只是一些收尾魔术;-)。dependents()
应该只有一个用于检索活动用户和资源的函数签名。另一方面,在编写代码时,我只想指定与路径操作函数相关的部分:资源和权限。剩下的就是如何让它发挥作用。开发和测试虚拟环境
fromfastapiimportDepends,FastAPIfromfastapi.securityimportOAuth2PasswordBearerfromfastapi_permissionsimportconfigure_permissions,Allow,DenyfrompydanticimportBaseModelapp=FastAPI()oauth2_scheme=OAuth2PasswordBearer(tokenUrl="/token")classItem(BaseModel):name:strowner:strdef__acl__(self):return[(Allow,Authenticated,"view"),(Allow,"role:admin","edit"),(Allow,f"user:{self.owner}","delete"),]classUser(BaseModel):name:strdefprincipals(self):return[f"user:{self.name}"]defget_current_user(token:str=Depends(oauth2_scheme)):...defget_active_user_principals(user:User=Depends(get_current_user)):...defget_item(item_identifier):...# Permission is already wrapped in Depends()Permission=configure_permissions(get_active_user_principals)@app.get("/item/{item_identifier}")asyncdefshow_item(item:Item=Permission("view",get_item)):return[{"item":item}]
1
fromfastapiimportDepends,FastAPIfromfastapi.securityimportOAuth2PasswordBearerfromfastapi_permissionsimportconfigure_permissions,Allow,DenyfrompydanticimportBaseModelapp=FastAPI()oauth2_scheme=OAuth2PasswordBearer(tokenUrl="/token")classItem(BaseModel):name:strowner:strdef__acl__(self):return[(Allow,Authenticated,"view"),(Allow,"role:admin","edit"),(Allow,f"user:{self.owner}","delete"),]classUser(BaseModel):name:strdefprincipals(self):return[f"user:{self.name}"]defget_current_user(token:str=Depends(oauth2_scheme)):...defget_active_user_principals(user:User=Depends(get_current_user)):...defget_item(item_identifier):...# Permission is already wrapped in Depends()Permission=configure_permissions(get_active_user_principals)@app.get("/item/{item_identifier}")asyncdefshow_item(item:Item=Permission("view",get_item)):return[{"item":item}]
2
make test
在本地测试任何更改。这将停止
在第一个错误时不报告覆盖范围。fromfastapiimportDepends,FastAPIfromfastapi.securityimportOAuth2PasswordBearerfromfastapi_permissionsimportconfigure_permissions,Allow,DenyfrompydanticimportBaseModelapp=FastAPI()oauth2_scheme=OAuth2PasswordBearer(tokenUrl="/token")classItem(BaseModel):name:strowner:strdef__acl__(self):return[(Allow,Authenticated,"view"),(Allow,"role:admin","edit"),(Allow,f"user:{self.owner}","delete"),]classUser(BaseModel):name:strdefprincipals(self):return[f"user:{self.name}"]defget_current_user(token:str=Depends(oauth2_scheme)):...defget_active_user_principals(user:User=Depends(get_current_user)):...defget_item(item_identifier):...# Permission is already wrapped in Depends()Permission=configure_permissions(get_active_user_principals)@app.get("/item/{item_identifier}")asyncdefshow_item(item:Item=Permission("view",get_item)):return[{"item":item}]
3
fromfastapiimportDepends,FastAPIfromfastapi.securityimportOAuth2PasswordBearerfromfastapi_permissionsimportconfigure_permissions,Allow,DenyfrompydanticimportBaseModelapp=FastAPI()oauth2_scheme=OAuth2PasswordBearer(tokenUrl="/token")classItem(BaseModel):name:strowner:strdef__acl__(self):return[(Allow,Authenticated,"view"),(Allow,"role:admin","edit"),(Allow,f"user:{self.owner}","delete"),]classUser(BaseModel):name:strdefprincipals(self):return[f"user:{self.name}"]defget_current_user(token:str=Depends(oauth2_scheme)):...defget_active_user_principals(user:User=Depends(get_current_user)):...defget_item(item_identifier):...# Permission is already wrapped in Depends()Permission=configure_permissions(get_active_user_principals)@app.get("/item/{item_identifier}")asyncdefshow_item(item:Item=Permission("view",get_item)):return[{"item":item}]
4
进行清洁
之前)fromfastapiimportDepends,FastAPIfromfastapi.securityimportOAuth2PasswordBearerfromfastapi_permissionsimportconfigure_permissions,Allow,DenyfrompydanticimportBaseModelapp=FastAPI()oauth2_scheme=OAuth2PasswordBearer(tokenUrl="/token")classItem(BaseModel):name:strowner:strdef__acl__(self):return[(Allow,Authenticated,"view"),(Allow,"role:admin","edit"),(Allow,f"user:{self.owner}","delete"),]classUser(BaseModel):name:strdefprincipals(self):return[f"user:{self.name}"]defget_current_user(token:str=Depends(oauth2_scheme)):...defget_active_user_principals(user:User=Depends(get_current_user)):...defget_item(item_identifier):...# Permission is already wrapped in Depends()Permission=configure_permissions(get_active_user_principals)@app.get("/item/{item_identifier}")asyncdefshow_item(item:Item=Permission("view",get_item)):return[{"item":item}]
5
推荐PyPI第三方库