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_permissionssource(请参见下面的进一步内容),然后启动测试服务器:

(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_permissions/example.py中用注释标记

为什么不使用作用域?

对于大多数应用程序,使用作用域来确定用户的权限就足够了。因此,如果作用域适合您的应用程序,请使用它们-它们已经是FastAPI框架的一部分。

虽然作用域仅与用户的状态相关联,但也有fastapi_权限 请考虑所请求资源的状态。

让我们以一篇科学论文为例:根据提交过程的状态(如"草稿"、"已提交"、"同行评审"或"已发布"),不同的用户应该具有不同的查看、编辑或收回权限。这可以与路径定义函数中的自定义代码一起实现,但fastapi权限提供了一种在单个位置定义这些约束的方法。

第二种情况是,FastAPI权限可能是您应用程序的正确附加功能:如果您的大脑像我的大脑一样连接/预处理到这样的权限模型-例如,长时间暴露在金字塔中../p>

长话短说:在您需要其他功能之前,请使用作用域

概念

由于fastapi_权限主要来自金字塔框架,如果您不清楚,我强烈建议查看它的安全文档。

系统依赖于fastapi中没有的几个概念:

  • 资源:提供访问控制列表的对象
  • 访问控制列表:定义哪些主体拥有哪些权限的规则列表
  • 主体:用户或其关联组/角色的标识符
  • 权限:对象上操作的标识符(字符串)

资源和访问控制列表

资源通过其\u acl\u属性提供访问控制列表。它可以是对象的属性,也可以是可调用的。列表中的每个条目都是一个包含三个值的元组:

  1. 一个操作:fastapi_permissions.allowfastapi_permissions.deny
  2. 负责人:例如"role:admin"或"user:bob"
  3. 权限或其元组:例如"edit"或("view"、"delete")
  4. < > >

    示例:

    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")]

    您不需要在访问控制列表的末尾添加任何"deny all子句",这是自动暗示的。acl中的所有条目都按列表中提供的顺序签入这使得一些复杂的配置变得简单,但有时也会使下背部疼痛…

    这两个原则EveryoneAuthenticated将在短时间内讨论。

    用户和主体标识符

    必须提供返回当前活动用户主体的函数。主体只是一个字符串列表,标识用户和用户所属的组/角色:

    示例:

    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

    特别负责人

    有两个特殊的主体也有助于提供访问控制列表:EveryoneAuthenticated

    应添加Everyone主体,而不考虑任何其他定义的主体或登录状态,Authenticated仅应为登录的用户添加。

    权限

    权限只是表示要对资源执行的操作的字符串。只要编点东西就行了。

    与特殊主体一样,有一个可用作通配符的特殊权限:fastapi\u permissions.all

    用法

    在使用权限系统之前,您必须提供以下内容:

    • 返回登录(活动)用户主体的可调用(fastapi dependencity
    • 具有访问控制列表的资源

    配置权限系统

    具有一些默认值的简单配置:

    fromfastapi_permissionsimportconfigure_permissions# must be provideddefget_active_principals(...):""" returns the principals of the current logged in user"""...permission=configure_permissions(get_active_principals)

    提供一个配置选项:

    • 权限例外:
      • 如果权限被拒绝,将引发此异常
      • 默认为FastAPI权限。权限异常
    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)

    在路径操作中使用权限

    若要在路径操作中使用访问控制,请使用权限和资源调用未配置的函数。如果授予了权限,则将返回选中该权限的请求资源,或者在本例中返回acl列表

    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?"。结果将是truefalse,因此不需要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签名,它在路径操作定义中用于搜索和注入依赖项。剩下的只是一些收尾魔术;-)。

    或者换句话说:要有一个好的api,path操作函数中的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

    开发需要安装flit:

    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

    欢迎加入QQ群-->: 979659372 Python中文网_新手群

    推荐PyPI第三方库


热门话题
JavaSpring重定向请求处理程序   SwingJava:拆分字符串并将其放入文本区域的   Java:标记“”上出现语法错误,此标记后面应为表达式   web服务Java RestService从日志文件写入和读取数据   java如何将ArrayList<String>转换为char数组,然后向后打印每个单词?   java SimpleDataFormat解析返回年终日期   加密Java aes解密bytebuffer,包括填充为空字节   java有没有办法从特定的if语句调用变量?   java从更新返回到渲染   spring GRPC Java登录测试   java为什么下面的代码不工作(StringBuffer.toString!=null)   java是一种可行的模式吗?   使用Spring集成测试的JavaOSGi片段   java jCommander为未知和未使用的值引发异常?   在imageView的editText中输入的java图像URL