超越登录：使用 ZITADEL 实施细粒度授权

介绍

当我们转向零信任思维时，传统 RBAC 系统等粗粒度安全措施的局限性就变得显而易见。向零信任转变的一个重要部分常常被忽视，那就是从粗粒度到细粒度安全的转变。

细粒度授权通过基于用户角色、操作甚至时间或位置等上下文等属性的访问来解决这个问题，而这种详细的访问控制对于现代应用程序至关重要。本文讨论如何齐塔德尔满足了这种细致入微的授权的需要。

借助 ZITADEL 的角色、元数据和操作等功能，用户可以获得适合零信任设置的高度详细的访问控制。此外，ZITADEL 可以与外部授权服务配合使用。

ZITADEL提供的授权机制

ZITADEL 是一个开源，用 Go 编写的云原生身份和访问管理解决方案 (IAM)。 ZITADEL 可作为 SaaS 解决方案使用，并且对于那些寻求自托管选项的人来说它也是开源的，从而确保了灵活性。它同时满足 B2C 和 B2B 用例。

其主要目标包括提供用于身份验证、授权、登录和单点登录 (SSO) 的统包功能，同时允许通过用户界面进行自定义。

它配备了广泛的审计跟踪来跟踪所有更改，使开发人员能够通过自定义代码（操作）扩展功能，支持广泛认可的标准，例如 OIDC、OAuth、SAML 和 LDAP，强调操作简便性和可扩展性，并为多功能集成。

基于角色的访问控制 (RBAC) 和委派访问

ZITADEL 使用 RBAC 来管理用户权限，其中权限与角色绑定，并且为用户分配这些角色。这简化了基于组织角色的用户访问管理。附加功能允许将角色委派给其他组织，从而促进与外部实体共享权限。

这对于相互关联或分层的组织尤其有价值。

虽然这些功能提供了强大的访问控制，但它们可能不足以满足复杂的授权需求，因此在 ZITADEL 中探索细粒度授权非常重要。

基于属性的访问控制 (ABAC) 的操作功能、自定义元数据和声明

ZITADEL通过引入动态的方式增强了传统的RBAC行动基于属性的访问控制 (ABAC) 功能。与根据用户角色授予访问权限的 RBAC 不同，ABAC 更加通用，可以在访问请求期间评估与用户、操作和资源相关的属性。

通过 ZITADEL 的操作，可以创建身份验证后脚本来分析特定的用户属性并在必要时阻止访问。

操作还可以建立自定义声明来增强 ABAC 系统，从而启用高级授权模型，根据位置、时间或任何可定义因素等属性来限制访问。

ZITADEL 允许管理员或获得许可的开发人员向用户和组织添加自定义元数据，从而扩大细粒度访问控制的可能性。

它通过从 CRM 或 HR 工具等外部系统收集额外数据来支持聚合索赔。 ZITADEL 还可以管理独特的资源，例如发货订单或 IoT 设备，并根据 User-Sub、角色、声明、IP 等属性确定访问权限。

扩展 ZITADEL 的现有细粒度访问控制功能

尽管 ZITADEL 具有全面的功能，但在某些情况下可能需要更定制或更细粒度的方法。

目前，在 ZITADEL 中实现细粒度授权的最有效方法是对较小的项目使用自定义应用程序逻辑，或者对于较大规模的项目，利用可用的第三方工具，例如grant.dev 、 cerbos.dev等。

这些工具可以与 ZITADEL 集成，进一步增强您进行细致入微、细粒度授权的能力。

一个实际的例子

假设一家媒体公司中有一个假设的新闻编辑室应用程序，它与后端 API 进行通信。记者用它来写作，而编辑则用它来编辑和发表这些文章。此 API 在本示例中是用 Python Flask 编写的，具有特定的端点，对这些端点的访问取决于用户的角色及其经验。端点：

• write_article ：仅供记者撰写。

• edit_article ：仅供编辑编辑文章。

• review_articles ：供高级记者和中高级编辑审阅文章。

• publish_article ：供中高级记者和高级编辑发布。在内部，API 使用 ZITADEL 发布的 JWT 来检查谁在发出请求。用户需要在请求标头中发送有效的 JWT。该JWT是在用户登录时获取的。

  
  

  JWT 包含有关用户的信息，例如他们的角色和经验。自定义声明中包含的此信息是此用例的关键。后端根据这些信息决定用户是否可以访问所请求的资源。

应用逻辑

• 用户入职：在用户入职过程中，每个用户都会获得一个角色，例如journalist或editor 。这是关键，因为它设置了谁在我们的设置中获得什么访问权限。管理经验/高级：除了角色之外，还跟踪用户的经验（例如我们示例中的junior 、 intermediate和senior ）。如果用户的体验发生变化，ZITADEL 会将其更新为元数据。如果用户加入 ZITADEL 时没有提及经验级别，系统就会假设其为“初级”。

• 用户登录：用户必须首先登录才能访问 API。成功登录后，ZITADEL 将返回一个包含用户信息的令牌。

• 令牌验证：当用户的请求访问 API 时，API 通过调用 ZITADEL 的令牌自省端点来验证令牌。尽管可以使用 JWKS 在本地验证 JWT，但我们还是使用 ZITADEL 的方法来检查令牌，以实现更好的安全性和即时令牌检查。通过这种方式，我们可以立即撤销代币，从一个地方管理它们，并且减少安全问题。它使我们的 API 登录和访问控制保持强大且与服务器保持同步。

• 细粒度访问控制：应用程序负责根据用户的角色和经验级别授权对资源的访问。它使用预定义的访问控制列表，将每个资源端点映射到有权访问它们的用户角色和经验级别。该列表用作授予或拒绝对资源的访问的规则手册。

• 关注点分离：在该 API 的设计中，特别注意确保业务逻辑和访问控制规则清晰分离。这对于应用程序的可维护性和可扩展性至关重要。通过将业务逻辑和访问规则分开，我们获得了更清晰的模块化设计。

  
  

  这使我们能够更新业务操作和访问规则，而不会相互影响。这提高了代码的可维护性，并且随着应用程序的扩展更容易管理。

  
  

  此外，这种设计使系统更加安全，因为访问规则从主要业务逻辑中抽象出来，降低了修改业务逻辑时意外引入安全漏洞的风险。

设置 ZITADEL

1. 创建 Media House 组织、新闻编辑室项目和文章 API

1. 创建 Media House 组织，转到“项目”，然后创建一个名为“Newsroom”的新项目。

   
   

   

   
   

2. 在 Newsroom 项目中，单击“新建”按钮创建一个新应用程序。

3. 添加名称，然后选择类型API 。

4. 选择基本作为身份验证方法，然后单击继续。

5. 现在检查您的配置，然后单击“创建” 。

6. 您现在将看到 API 的Client ID和Client Secret 。复制并保存它们。单击“关闭” 。

7. 当您单击左侧的URL时，您将看到相关的 OIDC URL。记下颁发者URL、 token_endpoint和introspection_endpoint 。

2. 在新闻编辑室项目中创建角色

1. 另外，记下您项目的资源 ID （转到项目并复制资源 ID）

2. 在项目仪表板上选择“在身份验证上声明角色”复选框，然后单击“保存” 。

3. 转至“角色” （从左侧菜单），然后单击“新建”以添加新角色。

4. 如下所示输入编辑和记者的角色，然后单击“保存” 。

5. 您现在将看到创建的角色。

3. 在 Newsroom 项目中创建用户

1. 转到组织中的“用户”选项卡（如下所示），然后转到“服务用户”选项卡。我们将在此演示中创建服务用户。要添加服务用户，请单击新建按钮。

   
   

   

   
   

2. 接下来，添加服务用户的详细信息，为访问令牌类型选择JWT ，然后单击创建。

   
   

   

   
   

3. 单击右上角的操作按钮。从下拉菜单中选择生成客户端密钥。

   
   

   

   
   

   
   

4. 复制您的客户端 ID 和客户端密钥。单击“关闭” 。

   
   

   

   
   

5. 现在，您拥有一个服务用户及其客户端凭据。

4. 为用户添加权限

1. 转到授权。单击新建。
   

   

   
   

2. 选择必须创建授权的用户和项目。单击继续。

   
   

   

   
   

3. 您可以在此处选择角色。为当前用户选择记者角色。单击“保存” 。

   
   

   

   
   

4. 您可以看到服务用户Lois Lane现在在Newsroom项目中具有记者角色。

5.向用户添加元数据

现在，让我们将元数据添加到用户配置文件中以指示他们的资历级别。使用“experience_level”作为键，并从“初级”、“中级”或“高级”中选择其值。

虽然我们通常可以假设此元数据是通过 HR 应用程序进行的 API 调用设置的，但为了简单和易于理解，我们将直接在控制台中设置元数据。

1. 转到元数据。单击编辑。
   

   
   

   

   
   

2. 提供experience_level作为键， senior作为值。单击保存图标，然后单击关闭按钮。

   
   

   

   
   

3. 用户现在拥有与其帐户关联的所需元数据。
   

   
   

   

   
   

4. 您还可以添加更多具有不同角色和 experience_levels（使用元数据）的服务用户，以使用前面的步骤测试演示。

   
   

   

6. 创建一个操作来捕获自定义声明中的角色和元数据

1. 单击“操作” 。单击“新建”以创建新操作。

2. 在创建操作部分中，为操作指定与函数名称相同的名称，即 allocateRoleAndExperienceClaims。在脚本字段中，复制/粘贴以下代码，然后单击“添加” 。

 function assignRoleAndExperienceClaims(ctx, api) { // Check if grants and metadata exist if (!ctx.v1.user.grants || !ctx.v1.claims['urn:zitadel:iam:user:metadata']) { return; } // Decode experience level from Base64 - metadata is Base64 encoded let experience_encoded = ctx.v1.claims['urn:zitadel:iam:user:metadata'].experience_level; let experience = ''; try { experience = decodeURIComponent(escape(String.fromCharCode.apply(null, experience_encoded.split('').map(function(c) { return '0x' + ('0' + c.charCodeAt(0).toString(16)).slice(-2); })))); } catch (e) { return; // If decoding fails, stop executing the function } // Check if the experience level exists if (!experience) { return; } // Iterate through the user's grants ctx.v1.user.grants.grants.forEach(grant => { // Iterate through the roles of each grant grant.roles.forEach(role => { // Check if the user is a journalist if (role === 'journalist') { // Set custom claims with the user's role and experience level api.v1.claims.setClaim('journalist:experience_level', experience); } // Check if the user is an editor else if (role === 'editor') { // Set custom claims with the user's role and experience level api.v1.claims.setClaim('editor:experience_level', experience); } }); }); }

3. allocateRoleAndExperienceClaims现在将被列为一个操作。

4. 接下来，我们必须选择Flow Type 。转到下面的“流程”部分。从下拉列表中选择补充标记。

5. 现在，您必须选择一个触发器。单击添加触发器。选择预访问令牌创建作为触发器类型，并选择allocateRoleAndExperienceClaims作为关联操作。

6. 现在，您将看到列出的触发器。

现在，当用户请求访问令牌时，将执行该操作，将用户角色和元数据转换为所需的格式，并将它们作为自定义声明添加到令牌中。然后，第三方应用程序可以使用此自定义声明来管理细粒度的用户访问。

设置 API 项目

从 GitHub 克隆项目：

运行以下命令以从此 GitHub 存储库克隆项目：

• git clone https://github.com/zitadel/example-fine-grained-authorization.git

导航到项目目录：

克隆后，导航到项目目录

• cd example-fine-grained-authorization 。

设置Python环境：

确保您已安装 Python 3 和 pip。您可以通过运行来检查这一点

• python3 --version和
• pip3 --version

在您的终端中。如果您没有安装 Python 或 pip，则需要安装它们。

接下来，通过运行为该项目创建一个新的虚拟环境

• python3 -m venv env 。

通过运行激活环境：

• 在 Windows 上： .\env\Scripts\activate
• 在 Unix 或 MacOS 上： source env/bin/activate

运行此命令后，您的终端应指示您现在正在 env 虚拟环境中工作。

安装依赖项：

使用位于项目目录（包含requirements.txt的目录）的终端，运行

• pip3 install -r requirements.txt

安装必要的依赖项。

配置环境变量：

该项目需要某些环境变量。使用我们从 ZITADEL 检索到的值填充.env文件。

 PROJECT_ID="<YOUR PROJECT ID>" ZITADEL_DOMAIN="<YOUR INSTANCE DOMAIN eg https://instance-as23uy.zitadel.cloud>" ZITADEL_TOKEN_URL="<YOUR TOKEN URL eg https://instance-as23uy.zitadel.cloud/oauth/v2/token" CLIENT_ID="<YOUR SERVICE USER'S CLIENT ID FROM THE GENERATED CLIENT CREDENTIALS eg sj_Alice>" CLIENT_SECRET="<YOUR SERVICE USER'S SECRET FROM THE GENERATED CLIENT CREDENTIALS"> ZITADEL_INTROSPECTION_URL="<YOUR INTROSPECTION URL eg https://instance-as23uy.zitadel.cloud/oauth/v2/introspect>" API_CLIENT_ID="<THE CLIENT ID OF YOUR API APPLICATION FOR BASIC AUTH eg 324545668690006737@api>" API_CLIENT_SECRET="<THE CLIENT SECRET OF YOUR API APPLICATION FOR BASIC AUTH>"

运行应用程序：

Flask API（在app.py中）使用 JWT 令牌和自定义声明来进行细粒度的访问控制。它会针对每个请求journalist和editor角色的自定义声明 experience_level，并使用此信息来决定经过身份验证的用户是否可以访问所请求的端点。

应用程序.py

 from flask import Flask, jsonify from auth import token_required from access_control import authorize_access app = Flask(__name__) # Define the /write_article route. @app.route('/write_article', methods=['POST']) @token_required def write_article(): authorization = authorize_access('write_article') if authorization is not True: return authorization # Resource-specific code goes here... return jsonify({"message": "Article written successfully!"}), 200 # Define the /edit_article route. @app.route('/edit_article', methods=['PUT']) @token_required def edit_article(): authorization = authorize_access('edit_article') if authorization is not True: return authorization # Resource-specific code goes here... return jsonify({"message": "Article edited successfully!"}), 200 # Define the /review_article route. @app.route('/review_articles', methods=['GET']) @token_required def review_article(): authorization = authorize_access('review_article') if authorization is not True: return authorization # Resource-specific code goes here... return jsonify({"message": "Article review accessed successfully!"}), 200 # Define the /publish_article route. @app.route('/publish_article', methods=['POST']) @token_required def publish_article(): authorization = authorize_access('publish_article') if authorization is not True: return authorization # Resource-specific code goes here... return jsonify({"message": "Article published successfully!"}), 200 # Add more endpoints as needed... if __name__ == '__main__': app.run(debug=True)

auth.py

 import os import jwt import requests from functools import wraps from flask import request, jsonify, g ZITADEL_INTROSPECTION_URL = os.getenv('ZITADEL_INTROSPECTION_URL') API_CLIENT_ID = os.getenv('API_CLIENT_ID') API_CLIENT_SECRET = os.getenv('API_CLIENT_SECRET') # This function checks the token introspection and populates the flask.g variable with the user's token def token_required(f): @wraps(f) def decorated(*args, **kwargs): token = request.headers.get('Authorization') if not token: abort(401) # Return status code 401 for Unauthorized if there's no token else: token = token.split(' ')[1] # The token is in the format "Bearer <token>", we want to extract the actual token # Call the introspection endpoint introspection_response = requests.post( ZITADEL_INTROSPECTION_URL, auth=(API_CLIENT_ID, API_CLIENT_SECRET), data={'token': token} ) if not introspection_response.json().get('active', False): return jsonify({"message": "Invalid token"}), 403 # Decode the token and print it for inspection decoded_token = jwt.decode(token, options={"verify_signature": False}) print(f"\n\n***** Decoded Token: {decoded_token} \n\n******") # Add the decoded token to Flask's global context g.token = decoded_token return f(*args, **kwargs) return decorated

access_control.py（模拟规则引擎的示例代码）

 import base64 import jwt from flask import g, jsonify # The access_requirements dictionary represents your access control rules. access_requirements = { 'write_article': [{'role': 'journalist', 'experience_level': 'junior'}, {'role': 'journalist', 'experience_level': 'intermediate'}, {'role': 'journalist', 'experience_level': 'senior'}], 'edit_article': [{'role': 'editor', 'experience_level': 'junior'}, {'role': 'editor', 'experience_level': 'intermediate'}, {'role': 'editor', 'experience_level': 'senior'}], 'review_articles': [{'role': 'journalist', 'experience_level': 'senior'}, {'role': 'editor', 'experience_level': 'intermediate'}, {'role': 'editor', 'experience_level': 'senior'}], 'publish_article': [{'role': 'journalist', 'experience_level': 'intermediate'}, {'role': 'journalist', 'experience_level': 'senior'}, {'role': 'editor', 'experience_level': 'senior'}] # Add more endpoints as needed... } # This function checks if the user is authorized to access the given endpoint. def authorize_access(endpoint): # We assume that the token has already been decoded in auth.py decoded_token = g.token # Initialize role and experience_level variables role = None experience_level = None for claim, value in decoded_token.items(): if ':experience_level' in claim: role, _ = claim.split(':') experience_level = base64.b64decode(value).decode('utf-8') break # If there's no role in the token, return an error if not role: return jsonify({"message": "Missing role"}), 403 # If there's a role in the token but no experience level, default the experience level to 'junior' if role and not experience_level: experience_level = 'junior' # If there's no role or experience level in the token, return an error if not role or not experience_level: return jsonify({"message": "Missing role or experience level"}), 403 # Get the requirements for the requested endpoint endpoint_requirements = access_requirements.get(endpoint) # If the endpoint is not in the access control list, return an error if not endpoint_requirements: return jsonify({"message": "Endpoint not found in access control list"}), 403 # Check if the user's role and experience level meet the requirements for the requested endpoint for requirement in endpoint_requirements: required_role = requirement['role'] required_experience_level = requirement['experience_level'] # Experience level hierarchy experience_levels = ['junior', 'intermediate', 'senior'] if role == required_role and experience_levels.index(experience_level) >= experience_levels.index(required_experience_level): return True #return jsonify({"message": "Access denied"}), 403 return jsonify({"message": f"Access denied! You are a {experience_level} {role} and therefore cannot access {endpoint}"}), 403

通过执行以下命令来运行 Flask 应用程序：

python3 app.py

如果一切设置正确，您的 Flask 应用程序现在应该正在运行。

该项目是使用 Python 3 开发和测试的，因此请确保您使用的是 Python 3 解释器。

运行并测试 API

运行API

1. 确保您已克隆存储库并安装了必要的依赖项，如前所述。

   
   

2. 运行client_credentials_token_generator.py脚本以生成访问令牌。

   
   

   client_credentials_token_generator.py

    import os import requests import base64 from dotenv import load_dotenv load_dotenv() ZITADEL_DOMAIN = os.getenv("ZITADEL_DOMAIN") CLIENT_ID = os.getenv("CLIENT_ID") CLIENT_SECRET = os.getenv("CLIENT_SECRET") ZITADEL_TOKEN_URL = os.getenv("ZITADEL_TOKEN_URL") PROJECT_ID = os.getenv("PROJECT_ID") # Encode the client ID and client secret in Base64 client_credentials = f"{CLIENT_ID}:{CLIENT_SECRET}".encode("utf-8") base64_client_credentials = base64.b64encode(client_credentials).decode("utf-8") # Request an OAuth token from ZITADEL headers = { "Content-Type": "application/x-www-form-urlencoded", "Authorization": f"Basic {base64_client_credentials}" } data = { "grant_type": "client_credentials", "scope": f"openid profile email urn:zitadel:iam:org:project:id:{PROJECT_ID}:aud urn:zitadel:iam:org:projects:roles urn:zitadel:iam:user:metadata" } response = requests.post(ZITADEL_TOKEN_URL, headers=headers, data=data) if response.status_code == 200: access_token = response.json()["access_token"] print(f"Response: {response.json()}") print(f"Access token: {access_token}") else: print(f"Error: {response.status_code} - {response.text}")

   
   

   打开终端并导航到项目目录，然后使用 python3 运行脚本：

   python3 client_credentials_token_generator.py

   
   

3. 如果成功，这将打印一个访问令牌到您的终端。这是您将用来验证您对 API 的请求的令牌。

   
   

4. 如果您之前没有启动 Flask API，请通过在项目目录中打开另一个终端并运行以下命令来运行 API：

   python3 app.py

   
   

5. API 服务器现在应该正在运行并准备好接受请求。

现在，您可以使用 cURL 或任何其他 HTTP 客户端（如 Postman）向 API 发出请求。请记住将curl命令中的your_access_token替换为您在步骤2中获得的访问令牌。

测试API

场景 1：初级编辑尝试编辑文章（成功）

具有editor角色和junior experience_level 的用户尝试调用edit_article端点。

• curl -H "Authorization: Bearer <your_access_token>" -X POST http://localhost:5000/edit_article

• 预期输出： {"message": "Article edited successfully"}

  
  

场景 2：初级编辑尝试发表文章（失败）

具有editor角色和junior experience_level 的用户尝试调用publish_article端点。

• curl -H "Authorization: Bearer <your_access_token>" -X POST http://localhost:5000/publish_article

• 预期输出： {"message": "Access denied! You are a junior editor and therefore cannot access publish_article"}

  
  

场景 3：资深记者尝试写一篇文章（成功）

具有journalist角色和senior experience_level 的用户尝试调用write_article端点。

• curl -H "Authorization: Bearer <your_access_token>" -X POST http://localhost:5000/write_article

• 预期输出： {"message": "Article written successfully"}

  
  

场景 4：初级记者尝试审阅文章（失败）

具有journalist角色和“初级”experience_level 的用户尝试调用review_articles端点。

• curl -H "Authorization: Bearer <your_access_token>" -X POST http://localhost:5000/review_articles

• 预期输出： {"message": "Access denied! You are a junior journalist and therefore cannot access review_articles"}

  
  

场景 5：高级编辑尝试审阅文章（成功）

具有editor角色和senior experience_level 的用户尝试访问review_articles端点。

• curl -H "Authorization: Bearer <your_access_token>" -X POST http://localhost:5000/review_articles

• 预期输出： {"message": "Article reviewed successfully"}

  
  

场景6：中级记者尝试发表文章（成功）

具有journalist角色和intermediate experience_level的用户尝试访问publish_article端点。

• curl -H "Authorization: Bearer <your_access_token>" -X POST http://localhost:5000/publish_article

• 预期输出： {"message": "Article published successfully"}

结论

在本文中，我们探讨了使用 ZITADEL 从传统 RBAC 转向更详细、更细粒度的授权方法的重要性。

我们深入研究了它的功能，例如 ABAC 的动态操作、与第三方工具集成的能力，并了解了如何将这些功能实际应用到现实场景中。

随着网络安全需求的增长，ZITADEL 等平台为复杂的授权挑战提供了必要的解决方案。