跳转到主要内容

使用Keycloak进行API端点保护的一个包

项目描述

ShieldAPI

shieldapi包为通过Keycloak-backend在Python Web应用程序(如FastAPI和Flask)中实现身份验证/授权机制提供实用工具。它包括管理用户账户、令牌身份验证和基于角色的访问控制的功能。

作者

安装

您可以使用pip安装此包

pip install shieldapi

先决条件

Keycloak

为了设置使用shieldapi包的Flask或FastAPI应用的Keycloak身份验证,必须满足一些先决条件

  1. 安装Keycloak - 您可以从官方Keycloak网站下载Keycloak。下载后,按照安装说明在您的机器上安装Keycloak。或者,您可以使用Docker镜像

  2. 创建领域 - 领域是一组用户、认证方法和客户端应用的容器。要在Keycloak中创建领域,请登录Keycloak管理控制台,点击“添加领域”,并填写必要的信息。有关创建领域的更多信息,请参阅官方Keycloak文档。

  3. 创建客户端 - 客户端是一个可以请求Keycloak的授权和/或认证服务的实体。要在Keycloak中创建客户端,请转到Keycloak管理控制台的“客户端”选项卡,点击“创建”,并填写必要的信息。有关创建客户端的更多信息,请参阅官方Keycloak文档。

  4. 创建客户端范围 - 客户端范围定义了一组可以授予用户的客户端级别角色。要在Keycloak中创建客户端范围,请转到Keycloak管理控制台的“客户端范围”选项卡,点击“创建”,并填写必要的信息。有关创建客户端范围的更多信息,请参阅官方Keycloak文档。

  5. 创建用户 - 用户是可以在Keycloak中认证并访问客户端应用的实体。要在Keycloak中创建用户,请转到Keycloak管理控制台的“用户”选项卡,点击“添加用户”,并填写必要的信息。有关创建用户的更多信息,请参阅官方Keycloak文档。

  6. 创建角色 - 角色是一组可以授予用户或客户端的权限集合。要在Keycloak中创建角色,请转到Keycloak管理控制台的“角色”选项卡,点击“添加角色”,并填写必要的信息。有关创建角色的更多信息,请参阅官方Keycloak文档。

  7. 将角色分配给客户端范围 - 为了授予用户或客户端权限,必须将角色分配给客户端范围。要在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_requiredhas_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

BasicLoginCredentialsHTTPBasicCredentials 的一个子类,它使用基本身份验证来验证用户,并通过 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。一旦令牌被验证,函数可以在返回客户端响应之前使用该令牌执行某些操作。

注意:BasicLoginCredentialsBasicLogin-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

BasicLoginHTTPBasic 的一个子类,它从 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]。这将

  1. 创建一个带有提升版本的标记版本,并将其推送到存储库。
  2. 触发 GitHub actions 工作流程,该工作流程创建 GitHub 版本并发布到 PyPI。

附加说明

  • 本项目遵循语义版本控制。
  • 使用 --dry 选项来预览发布更改。

致谢

这项工作得到由 Horizon 2020 资助的 MarketPlace 项目的支持,该项目是在 H2020-NMBP-25-2017 号征集下进行的(资助号 760173)。

许可证

代码采用 BSD-3-Clause 许可。版权所有 © 2023 Materials MarketPlace

项目详情


下载文件

下载适合您平台的文件。如果您不确定选择哪个,请了解更多关于 安装包 的信息。

源分布

shieldapi-1.2.0.tar.gz (20.2 kB 查看散列值)

上传时间

构建分布

shieldapi-1.2.0-py3-none-any.whl (17.8 kB 查看散列值)

上传时间 Python 3

由...