使用Keycloak进行API端点保护的一个包
项目描述
ShieldAPI
该shieldapi
包为通过Keycloak-backend在Python Web应用程序(如FastAPI和Flask)中实现身份验证/授权机制提供实用工具。它包括管理用户账户、令牌身份验证和基于角色的访问控制的功能。
作者
- Yoav Nahshon (yoav.nahshon@iwm.fraunhofer.de)
- Pablo De Andres (pablo.de.andres@iwm.fraunhofer.de)
- Matthias Büschelberger (matthias.bueschelberger@iwm.fraunhofer.de)
- Treesa Joseph (treesa.joseph@sintef.no)
- Thomas Hagelien (thomas.f.hagelien@sintef.no)
- Simon Adorf
安装
您可以使用pip安装此包
pip install shieldapi
先决条件
Keycloak
为了设置使用shieldapi包的Flask或FastAPI应用的Keycloak身份验证,必须满足一些先决条件
-
安装Keycloak - 您可以从官方Keycloak网站下载Keycloak。下载后,按照安装说明在您的机器上安装Keycloak。或者,您可以使用Docker镜像。
-
创建领域 - 领域是一组用户、认证方法和客户端应用的容器。要在Keycloak中创建领域,请登录Keycloak管理控制台,点击“添加领域”,并填写必要的信息。有关创建领域的更多信息,请参阅官方Keycloak文档。
-
创建客户端 - 客户端是一个可以请求Keycloak的授权和/或认证服务的实体。要在Keycloak中创建客户端,请转到Keycloak管理控制台的“客户端”选项卡,点击“创建”,并填写必要的信息。有关创建客户端的更多信息,请参阅官方Keycloak文档。
-
创建客户端范围 - 客户端范围定义了一组可以授予用户的客户端级别角色。要在Keycloak中创建客户端范围,请转到Keycloak管理控制台的“客户端范围”选项卡,点击“创建”,并填写必要的信息。有关创建客户端范围的更多信息,请参阅官方Keycloak文档。
-
创建用户 - 用户是可以在Keycloak中认证并访问客户端应用的实体。要在Keycloak中创建用户,请转到Keycloak管理控制台的“用户”选项卡,点击“添加用户”,并填写必要的信息。有关创建用户的更多信息,请参阅官方Keycloak文档。
-
创建角色 - 角色是一组可以授予用户或客户端的权限集合。要在Keycloak中创建角色,请转到Keycloak管理控制台的“角色”选项卡,点击“添加角色”,并填写必要的信息。有关创建角色的更多信息,请参阅官方Keycloak文档。
-
将角色分配给客户端范围 - 为了授予用户或客户端权限,必须将角色分配给客户端范围。要在Keycloak中将角色分配给客户端范围,请转到Keycloak管理控制台的“客户端范围”选项卡,点击您在步骤4中创建的客户端范围,然后点击“映射”选项卡。从这里,您可以分配角色到客户端范围。
环境变量
为了成功启用令牌生成和令牌验证的回调,请确保以下环境变量设置正确。
变量名 | 描述 | 示例值 |
---|---|---|
KEYCLOAK_HOST |
Keycloak实例的主机名 | "http://localhost:8080" |
KEYCLOAK_REALM_NAME |
在前一章步骤2中设置的领域名称 | "my_realm" |
KEYCLOAK_CLIENT_ID |
在前一章步骤3中创建的客户端ID。 | "shieldapi" |
KEYCLOAK_CLIENT_SECRET |
在前一章步骤3中创建的客户端密钥(应将access_type 设置为“confidential”)。 |
"1169a83e-a237-40ad-90db-8b8d4848de09" |
KEYCLOAK_VERIFY_HOST |
控制是否对HTTPS请求的SSL证书进行验证。默认设置为False 。 |
"True" |
KEYCLOAK_REALM_ADMIN_USER |
领域管理员用户名,如果需要在Flask/FastAPI应用中以管理员身份执行任何操作。 | "admin" |
KEYCLOAK_REALM_ADMIN_PASSWORD |
领域管理员密码,如果需要在Flask/FastAPI应用中以管理员身份执行任何操作。 | "admin_password" |
与Flask一起使用
有几个装饰器可以包含到Flask应用中,以进行端点的认证。
以下列出了一些示例
login_required
要使用login_required
装饰器,只需将装饰器添加到任何需要认证的Flask路由函数上。装饰器通过检查访问令牌来检查用户是否已登录,如果未找到或过期,则使用401未授权错误终止请求。如果访问令牌有效,则正常执行路由函数。
from flask import Flask
from shieldapi.frameworks.flask import login_required
app = Flask(__name__)
@app.route('/protected_endpoint')
@login_required
def hello_world():
return 'Hello, World!'
if __name__ == '__main__':
app.run()
使用WSGI服务器运行flask应用(推荐用于生产)。
如果服务正在运行,则作为客户端访问服务的端点
import requests
ACCESS_TOKEN = "your_access_token_here"
URL = "http://localhost:5000/protected_endpoint"
headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
# GET request to the root endpoint
response = requests.get(URL, headers=headers)
print(response.text) # prints 'Hello, World!'
admin_required
这是一个Flask装饰器,用于检查用户是否为管理员。装饰器从请求头中提取访问令牌,并验证令牌是否有效且处于活动状态。然后检查用户在令牌的“roles”声明中是否具有“admin”角色。如果任何检查失败,装饰器将引发401未授权或403禁止访问错误。
示例
from flask import Flask
from shieldapi.frameworks.flask import admin_required
app = Flask(__name__)
@app.route('/')
def index():
return 'Home Page'
@app.route('/admin')
@admin_required
def admin_view():
return 'Admin Dashboard'
if __name__ == '__main__':
app.run(debug=True)
在此示例中,admin_view()
函数只能由在其访问令牌中具有"admin
"角色的用户访问。如果未具有"admin
"角色的用户尝试访问视图,将引发403禁止访问错误。
has_scope
has_scope
函数创建一个装饰器,该装饰器限制对已被授予所需范围的应用的访问。装饰器接收所需范围的列表,并检查用户的访问令牌是否包含所有这些范围。如果没有,用户将被拒绝访问。
要使用此装饰器,您只需将@has_scope
装饰器添加到要基于范围限制访问的路由函数上方即可。装饰器以所需范围的列表作为参数。
例如
from flask import Flask
from shieldapi.frameworks.flask import has_scope
app = Flask(__name__)
@app.route('/admin')
@has_scope(['admin'])
def admin():
# code to handle admin requests
在此示例中,admin()
函数仅对具有在访问令牌中授予"admin
"范围的用户的可访问。如果用户的访问令牌不包括"admin
"范围,装饰器中的受限函数将返回None,用户将被拒绝访问
get_userinfo
get_userinfo
函数用于在Flask应用程序中从访问令牌中检索用户信息。
要使用此函数,您应首先获取访问令牌,通常通过使用如admin_required
或has_scope
之类的装饰器,这些装饰器将装饰需要用户信息的路由或方法。
获取访问令牌后,您可以调用get_userinfo
函数从令牌中检索用户信息。
该函数返回一个包含用户信息的字典,例如用户的电子邮件、姓名和其他包含在访问令牌中的属性。
与FastAPI的用法
该包提供了几个FastAPI依赖类。
以下列出了一些示例
AuthTokenBearer
要在FastAPI应用程序中使用AuthTokenBearer
HTTPBearer子类,您可以将其添加为需要身份验证的API路由函数的依赖项。以下是如何使用它的示例:
from fastapi import FastAPI, Depends
from shieldapi.frameworks.fastapi import AuthTokenBearer
app = FastAPI()
# Create an instance of the AuthTokenBearer class
auth_bearer = AuthTokenBearer()
# Define a protected route that requires authentication
@app.get("/protected_endpoint")
async def protected_route(auth: str = Depends(auth_bearer)):
# Use the authentication token to access protected resources
return {"message": "This resource is protected"}
在此示例中,将AuthTokenBearer
实例auth_bearer添加到受保护的函数protected_route中作为依赖项。这确保了路由受到保护,并且只能由经过身份验证的用户访问。
当对protected_route发出请求时,FastAPI将调用AuthTokenBearer
实例的__call__
方法以验证并从请求头中提取身份验证令牌。如果令牌缺失或已过期,将引发HTTPException。
要使用与Keycloak身份验证一起的AuthTokenBearer
子类,您需要在shieldapi.keycloak_utils
模块中的get_keycloak_openid()
函数中配置您的Keycloak服务器URL和域名称。此外,您还需要在应用程序中定义适当的Keycloak角色和权限,以控制对受保护资源的访问。
请注意,此子类假设身份验证令牌按"Bearer {token}
"格式在请求头中传递。如果您的应用程序使用不同的格式,您可能需要修改__call__
方法以正确格式提取令牌。
对于示例用法,使用WSGI服务器(推荐用于生产)或TestClient
(仅推荐用于生产)运行FastAPI应用程序。
如果服务正在运行,则作为客户端访问服务的端点
import requests
ACCESS_TOKEN = "your_access_token_here"
URL = "http://localhost:8080/protected_endpoint"
headers = {"Authorization": f"Bearer {ACCESS_TOKEN}"}
# GET request to the root endpoint
response = requests.get(URL, headers=headers)
print(response.json()) # prints: {"message": "This resource is protected"}
depends_auth_token_bearer
除了上面的AuthTokenBearer
之外,还可以使用辅助函数depends_auth_token_bearer
。
from fastapi import FastAPI, Depends
from shieldapi.frameworks.fastapi import depends_auth_token_bearer
app = FastAPI()
# Define a protected route that requires authentication
@app.get("/protected_endpoint")
async def protected_route(auth: str = Depends(depends_auth_token_bearer)):
# Use the authentication token to access protected resources
return {"message": "This resource is protected"}
BasicLoginCredentials
BasicLoginCredentials
是 HTTPBasicCredentials
的一个子类,它使用基本身份验证来验证用户,并通过 login
函数从 Keycloak 获取访问令牌。如果身份验证凭据有效,它将以 "Bearer {token}"
格式返回身份验证令牌作为字符串。
以下是使用 BasicLoginCredentials
在 FastAPI 中进行身份验证的示例:
from fastapi import FastAPI, Depends, HTTPException
from fastapi.security import HTTPBasic
from shieldapi.frameworks.fastapi import BasicLoginCredentials
app = FastAPI()
security = HTTPBasic()
@app.get("/protected_endpoint")
async def protected_route(login: BasicLoginCredentials = Depends(security)):
"""
This route is protected and authentication with user/password in order
to generate a token.
"""
if not login.token:
raise HTTPException(
status_code=401,
detail="Invalid authentication credentials"
)
# Perform some action with the token here
return {"message": "Authenticated successfully!"}
在上面的示例中,protected_endpoint
被用作 BasicLoginCredentials
依赖项进行保护。当对这条路由发出请求时,FastAPI
将提取带有基本身份验证凭据的 Authorization 标头。然后,BasicLoginCredentials
依赖项将使用这些凭据对 Keycloak 进行身份验证,并以 "Bearer {token}"
格式的字符串形式返回身份验证令牌。如果身份验证凭据无效,它将引发一个状态码为 401、消息为 "Invalid authentication credentials"
的 HTTPException
。一旦令牌被验证,函数可以在返回客户端响应之前使用该令牌执行某些操作。
注意:BasicLoginCredentials 和 BasicLogin-dependency 之间的主要区别在于,后者需要在将其作为依赖项注入端点或应用程序之前进行实例化。BasicLogin 简单地返回包含令牌的
str
,而 BasicLoginCredentials 返回包含用户名和密码的完整请求对象。
为了演示使用,请使用 WSGI 服务器(推荐用于生产)或 TestClient
(仅推荐用于生产)运行 FstAPI 应用程序。
如果服务正在运行,则作为客户端访问服务的端点
import requests
URL = "http://localhost:8080/protected_endpoint"
username = "your_username"
password = "your_password"
response = requests.get(URL, auth=(username, password))
print(response.json()) # prints: {"message": "Authenticated successfully!"}
BasicLogin
BasicLogin
是 HTTPBasic
的一个子类,它从 HTTP 基本身份验证头中提取用户名和密码,并使用它们通过登录函数从 Keycloak 获取访问令牌。然后使用返回的访问令牌进行请求身份验证。
示例
from fastapi import FastAPI, Depends
from shieldapi.frameworks.fastapi import BasicLogin
app = FastAPI()
security = BasicLogin()
@app.get("/protected_endpoint")
async def get_user(token: str = Depends(security)):
"""
This route is protected and authentication with user/password in order
to generate a token.
"""
if not token:
raise HTTPException(
status_code=401,
detail="Invalid authentication credentials"
)
# Perform some action with the token here
return {"message": "Authenticated successfully!"}
在此示例中,BasicLogin
被用作 protected_endpoint
函数的依赖项。当调用该函数时,BasicLogin
将从请求头中提取基本认证凭据,将它们传递给登录函数以从 Keycloak 获取访问令牌,并以 "Bearer {token}"
格式的字符串返回访问令牌。受保护的函数的 token 参数将包含 "Bearer {token}"
格式的身份验证令牌作为字符串。
为了演示使用,请使用 WSGI 服务器(推荐用于生产)或 TestClient
(仅推荐用于生产)运行 FstAPI 应用程序。
如果服务正在运行,则作为客户端访问服务的端点
import requests
URL = "http://localhost:8080/protected_endpoint"
username = "your_username"
password = "your_password"
response = requests.get(URL, auth=(username, password))
print(response.json()) # prints: {"message": "Authenticated successfully!"}
depends_basic_login
除了上面的 BasicLogin
之外,还可以使用辅助函数 depends_basic_login
。
from fastapi import FastAPI, Depends
from shieldapi.frameworks.fastapi import depends_basic_login
app = FastAPI()
# Define a protected route that requires authentication
@app.get("/protected_endpoint")
async def protected_route(auth: str = Depends(depends_basic_login)):
# Use the authentication token to access protected resources
return {"message": "This resource is protected"}
贡献
以问题、评论和拉取请求形式提供的贡献非常欢迎。
要做出代码贡献,请将此存储库分支出来,然后创建一个拉取请求。对于开发,您需要设置一个 Python 环境(带有最新的 Python 版本),安装开发需求,并使用以下命令安装预提交钩子:
pip install .[dev]
pre-commit install
测试
此存储库的测试是通过 pytest 实现的。要运行这些测试,首先使用以下命令安装测试依赖项:
pip install '.[tests]'
然后使用 pytest
命令运行测试。
对于维护者
要创建一个新版本,请克隆存储库,使用 pip install -e '.[dev]'
安装开发依赖项,然后执行 bumpver update --[major|minor|patch]
。这将
- 创建一个带有提升版本的标记版本,并将其推送到存储库。
- 触发 GitHub actions 工作流程,该工作流程创建 GitHub 版本并发布到 PyPI。
附加说明
- 本项目遵循语义版本控制。
- 使用
--dry
选项来预览发布更改。
致谢
这项工作得到由 Horizon 2020 资助的 MarketPlace 项目的支持,该项目是在 H2020-NMBP-25-2017 号征集下进行的(资助号 760173)。
许可证
代码采用 BSD-3-Clause 许可。版权所有 © 2023 Materials MarketPlace
项目详情
下载文件
下载适合您平台的文件。如果您不确定选择哪个,请了解更多关于 安装包 的信息。