Beyond Login: Implement Fine-Grained Authorization With ZITADEL

Introduction

As we move towards a zero-trust mindset, the limitation of coarse-grained security measures like the traditional RBAC system becomes clear. An essential part of the shift to zero trust that often goes undiscussed is the move from coarse-grained to fine-grained security.

Fine-grained authorization addresses this by basing access on attributes like user roles, actions, and even context like time or location, and such detailed access control is vital for modern applications. This article discusses how ZITADEL caters to the need for such nuanced authorization.

With ZITADEL's features like roles, meta-data, and actions, users can obtain highly detailed access control suited for a zero-trust setting. Additionally, ZITADEL can work with external authorization services.

Authorization Mechanisms Offered by ZITADEL

ZITADEL is an open-source, cloud-native Identity and Access Management solution (IAM) written in Go. ZITADEL is available as a SaaS solution, and it's also open source for those seeking self-hosting options, ensuring flexibility. It caters to both B2C and B2B use cases.

Its primary objectives include providing turnkey features for authentication, authorization, login, and single sign-on (SSO) while allowing customization through user interfaces.

It comes with an extensive audit trail for tracking all changes, enables developers to extend functionalities with custom code (actions), supports widely recognized standards like OIDC, OAuth, SAML, and LDAP, emphasizes ease of operation and scalability, and offers comprehensive APIs for versatile integration.

Role-Based Access Control (RBAC) and Delegated Access

ZITADEL uses RBAC to manage user permissions, where permissions are tied to roles, and users are allocated these roles. This simplifies user access management based on their organizational roles. An additional feature allows roles to be delegated to other organizations, facilitating permissions sharing with external entities.

This is especially valuable for interconnected or hierarchical organizations.

While these capabilities offer robust access control, they might not be enough for intricate authorization needs, hence the importance of exploring fine-grained authorization in ZITADEL.

The Actions Feature, Custom Metadata, and Claims for Attribute-based Access Control (ABAC)

ZITADEL enhances the traditional RBAC by introducing its dynamic actions feature for attribute-based access control (ABAC). Unlike RBAC, which grants access based on user roles, ABAC is more versatile, assessing attributes linked to the user, action, and resource during access requests.

With ZITADEL's actions, post-authentication scripts can be created to analyze specific user attributes and block access when necessary.

Actions can also establish custom claims to boost the ABAC system, enabling advanced authorization models that restrict access based on attributes like location, time, or any definable factor.

ZITADEL lets administrators or permitted developers add custom metadata to users and organizations, amplifying fine-grained access control possibilities.

It supports aggregated claims by gathering extra data from external systems like CRM or HR tools. ZITADEL can also manage unique resources, such as shipping orders or IoT devices, and determine access based on attributes like User-Sub, Roles, Claims, IP, and more.

Extending ZITADEL's Existing Capabilities for Fine-Grained Access Control

Despite the comprehensive features that come with ZITADEL, there may be instances where a more customized or fine-grained approach is needed.

Currently, the most effective way to implement fine-grained authorization in ZITADEL is by using custom application logic for smaller projects, or for larger scale projects, leveraging an available third-party tool such as warrant.dev, cerbos.dev, etc.

These tools can integrate with ZITADEL, further enhancing your capacity for nuanced, fine-grained authorization.

A Practical Example

Let's say there's a hypothetical Newsroom Application in a media company, which talks to a back-end API. Journalists use it to write, while editors edit and publish these articles. This API, written in Python Flask in this example, has specific endpoints and access to these endpoints depends on the user's role and how experienced they are. The endpoints:

write_article: Only for journalists to write.

edit_article: Just for editors to edit articles.

review_articles: For senior journalists and intermediate and senior editors to review articles.

publish_article: For intermediate and senior journalists and senior editors to publish. Internally, the API uses a JWT issued by ZITADEL for checking who's making requests. Users need to send a valid JWT in their request's header. This JWT was obtained when the user logged in.

The JWT contains info about the user, like their role and experience. This info, contained within custom claims, is key to this use case. The backend decides if the user can access the requested resource based on this information.

The Application Logic

User Onboarding: During the user onboarding process, each user gets a role, e.g., journalist or editor. This is key since it sets who gets what access in our setup. Managing Experience/Seniority: Besides roles, a user's experience (like junior, intermediate, and senior in our example) is tracked. If a user's experience changes, ZITADEL updates it as metadata. If there's no experience level mentioned when a user onboards ZITADEL, the system just assumes it's 'junior.'

User Login: A user must first login to access the API. Upon successful login, ZITADEL returns a token with the user’s information.

Token Validation: When a request from a user hits the API, the API validates the token by calling ZITADEL's token introspection endpoint. Although JWTs can be validated locally using JWKS, we went with ZITADEL's method to inspect tokens for better security and instant token checks. This way, we can revoke tokens instantly, manage them from one place, and have fewer security issues. It keeps our API's login and access controls strong and up-to-date with the server.

Fine-Grained Access Control: The application is responsible for authorizing access to resources based on a user's role and experience level. It uses a predefined access control list that maps each resource endpoint to the user roles and experience levels authorized to access them. This list serves as the rulebook for granting or denying access to resources.

Separation of Concerns: In the design of this API, special attention was given to ensuring that business logic and access control rules are cleanly separated. This is crucial for the maintainability and scalability of the application. By keeping business logic and access rules separate, we get a cleaner, modular design.

This lets us update business actions and access rules without affecting each other. This increases the maintainability of the code and makes it easier to manage as the application scales.
Additionally, this design makes the system more secure as access rules are abstracted away from the main business logic, reducing the risk of accidentally introducing security vulnerabilities when modifying the business logic.

Set up ZITADEL

1. Create Media House Organization, Newsroom Project and Article API

Create the Media House organization, go to Projects, and create a new project called Newsroom.
In the Newsroom project, click the New button to create a new application.

Add a name, and select type API.

Select Basic as the authentication method, and click Continue.

Now review your configuration, and click Create.

You will now see the API’s Client ID and the Client Secret. Copy them and save them. Click Close.

When you click URLs on the left, you will see the relevant OIDC URLs. Note down the issuer URL, token_endpoint, and introspection_endpoint.

2. Create Roles in the Newsroom Project

Also, note down the Resource ID of your project (go to the project and copy the Resource ID)

Select the Assert Roles on the Authentication checkbox on the project dashboard, and click Save.

Go to Roles (from the left menu), and click New to add new roles.

Enter the roles of editor and journalist as shown below, and click Save.

You will now see the created roles.

3. Create Users in the Newsroom Project

Go to the Users tab in your organization as shown below, and go to the Service Users tab. We will be creating service users in this demo. To add a service user, click the New button.
Next, add the details of the service user, select JWT for Access Token Type, and click Create.
Click the Actions button on the top right corner. Select Generate Client Secret from the drop-down menu.
Copy your Client ID and Client Secret. Click Close.
Now, you have a service user, along with their client credentials.

4. Add Authorizations for the Users

Go to Authorizations. Click New.
Select the user and the project for which the authorization must be created. Click Continue.
You can select a role here. Select the role of journalist for the current user. Click Save.
You can see the service user Lois Lane now has the role of journalist in the Newsroom project.

5. Add Metadata to the Users

Now, let's add metadata to the user profile to indicate their level of seniority. Use 'experience_level' as the key, and for its value, choose from 'junior', 'intermediate', or 'senior.'

Although we can typically assume this metadata is set through an API call made by the HR application, for simplicity and ease of understanding, we will set the metadata directly in the console.

Go to Metadata. Click Edit.
Provide experience_level as the key and senior as the value. Click the save icon, and click the Close button.
The user now has the required metadata associated with their account.
You can also add a few more service users with different roles and experience_levels (using metadata) to test the demo using the previous steps.

6. Create an Action to Capture Role and Metadata in Custom Claim

1. Click on Actions. Click New to create a new action.

2. In the Create an Action section, give the action the same name as the function name, i.e., assignRoleAndExperienceClaims. In the script field, copy/paste the following code, and then click Add.

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);
            }
        });
    });
}

The assignRoleAndExperienceClaims will now be listed as an action.

Next, we must select a Flow Type. Go to the Flows section below. Select Complement Token from the dropdown.

Now, you must choose a trigger. Click Add trigger. Select pre-access token creation as the trigger type and select assignRoleAndExperienceClaims as the associated action.

And now, you will see the trigger listed.

Now, when a user requests an access token, the action will be executed, transforming the user roles and metadata into the required format and adding them as a custom claim to the token. This custom claim can then be used by third-party applications to manage fine-grained user access.

Set Up the API Project

Clone the Project from GitHub:

Run the command below to clone the project from this GitHub repository:

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

Navigate to the Project Directory:

After cloning, navigate to the project directory with

cd example-fine-grained-authorization.

Setup a Python Environment:

Ensure you have Python 3 and pip installed. You can check this by running

python3 --version and
pip3 --version

in your terminal. If you don't have Python or pip installed, you will need to install them.

Next, create a new virtual environment for this project by running

python3 -m venv env.

Activate the environment by running:

On Windows: .\env\Scripts\activate
On Unix or MacOS: source env/bin/activate

After running this command, your terminal should indicate that you are now working inside the env virtual environment.

Install Dependencies:

With the terminal at the project directory (the one containing requirements.txt), run

pip3 install -r requirements.txt

to install the necessary dependencies.

Configure Environment Variables:

The project requires certain environment variables. Fill in the .env file with the values we retrieved from ZITADEL.

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

Run the Application:

The Flask API (in app.py) uses JWT tokens and custom claims for fine-grained access control. It checks the custom claim experience_level for the roles journalist and editor on every request, using this information to decide if the authenticated user can access the requested endpoint.

app.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 (sample code simulating a rules engine)

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

Run the Flask application by executing:

python3 app.py

If everything is set up correctly, your Flask application should now be running.

This project was developed and tested with Python 3, so make sure you are using a Python 3 interpreter.

Run and Test the API

Run the API

Ensure you have cloned the repository and installed the necessary dependencies as described earlier.

Run the client_credentials_token_generator.py script to generate an access token.

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

Open your terminal and navigate to the project directory, then run the script using python3:

python3 client_credentials_token_generator.py

If successful, this will print an access token to your terminal. This is the token you will use to authenticate your requests to the API.
If you didn't start the Flask API earlier, run the API by opening another terminal in the project directory and running:

python3 app.py
The API server should be now running and ready to accept requests.

Now, you can use cURL or any other HTTP client (like Postman) to make requests to the API. Remember to replace your_access_token in the curl commands with the access token you obtained in step 2.

Test the API

Scenario 1: Junior Editor Tries to Edit an Article (Success)

User with editor role and junior experience_level tries to call edit_article endpoint.

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

Expected Output: {"message": "Article edited successfully"}

Scenario 2: Junior Editor Tries to Publish an Article (Failure)

User with editor role and junior experience_level tries to call publish_article endpoint.

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

Expected output: {"message": "Access denied! You are a junior editor and therefore cannot access publish_article"}

Scenario 3: Senior Journalist Tries to Write an Article (Success)

User with journalist role and senior experience_level tries to call write_article endpoint.

curl -H "Authorization: Bearer <your_access_token>" -X POST http://localhost:5000/write_article
Expected output: {"message": "Article written successfully"}

Scenario 4: Junior Journalist Tries to Review Articles (Failure)

User with journalist role and 'junior' experience_level tries to call review_articles endpoint.

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

Expected output: {"message": "Access denied! You are a junior journalist and therefore cannot access review_articles"}

Scenario 5: Senior Editor Tries to Review Articles (Success)

User with editor role and senior experience_level tries to access review_articles endpoint.

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

Expected output: {"message": "Article reviewed successfully"}

Scenario 6: Intermediate Journalist Tries to Publish an Article (Success)

User with journalist role and intermediate experience_level tries to access publish_article endpoint.

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

Expected output: {"message": "Article published successfully"}

Conclusion

In this article, we explored the importance of shifting from traditional RBAC to a more detailed, fine-grained authorization approach using ZITADEL.

We delved into its features like dynamic actions for ABAC, the ability to integrate with third-party tools, and saw how these capabilities can be practically applied in a real-world scenario.

As the demands of cybersecurity grow, platforms like ZITADEL provide the necessary solutions for complex authorization challenges.