mirror of
https://github.com/zitadel/zitadel.git
synced 2026-08-17 16:35:14 -05:00
Wim Van Laer
699ed17918
# Which Problems Are Solved naming inconsistencies: Configuration <-> Settings # How the Problems Are Solved # Additional Changes # Additional Context - Closes #11282
256 lines
9.8 KiB
Plaintext
256 lines
9.8 KiB
Plaintext
---
|
|
title: ZITADEL with Python
|
|
sidebar_label: Python
|
|
---
|
|
|
|
This example shows you how to secure a Python3 Flask API with both authentication and authorization using ZITADEL.
|
|
|
|
## Overview
|
|
|
|
|
|
|
|
The Python API will have public, private, and private-scoped routes and check if a user is authenticated and authorized to access the routes.
|
|
The private routes expect an authorization header with a valid access token in the request. The access token is used as a bearer token to authenticate the user when calling the API.
|
|
The API will validate the access token on the [introspect endpoint](/apis/openidoauth/endpoints#introspection-endpoint) and will receive the user's roles from ZITADEL.
|
|
|
|
The API application uses [Client Secret Basic](/apis/openidoauth/authn-methods#client-secret-basic) to authenticate against ZITADEL and access the introspection endpoint.
|
|
You can use any valid access_token from a user or service account to send requests to the example API.
|
|
In this example we will use a service account with a [personal access token](/guides/integrate/service-accounts/personal-access-token) which can be used directly to access the example API.
|
|
|
|
## Running the example
|
|
|
|
### Python Prerequisites
|
|
|
|
In order to run the example you need to have `python3` and `pip3` installed.
|
|
|
|
### ZITADEL settings for the API
|
|
|
|
|
|
|
|
You need to setup a couple of things in ZITADEL.
|
|
|
|
1. If you don't have an instance yet, please go ahead and create an instance as explained [here](/guides/start/quickstart#3-create-your-zitadel-instance). Also, create a new project by following the steps [here](/guides/start/quickstart#4-create-your-project-and-application).
|
|
|
|
2. You must create an API application in your project. Follow [this guide](/guides/manage/console/applications-overview) to create a new application of type "API" with authentication method "Basic". Save both the ClientID and ClientSecret after you create the application.
|
|
|
|
### Create the API
|
|
|
|
1. Clone or download this [Python project](https://github.com/zitadel/example-api-python3-flask) to your workspace.
|
|
|
|
```
|
|
git clone https://github.com/zitadel/example-api-python3-flask
|
|
|
|
cd example-api-python3-flask
|
|
```
|
|
2. The [server.py](https://github.com/zitadel/example-api-python3-flask/blob/main/server.py) file contains a Flask-based API that provides authentication for routes using the OpenID Connect protocol as shown below.
|
|
|
|
```python
|
|
from flask import Flask, jsonify, Response
|
|
from authlib.integrations.flask_oauth2 import ResourceProtector
|
|
from validator import ZitadelIntrospectTokenValidator, ValidatorError
|
|
|
|
require_auth = ResourceProtector()
|
|
require_auth.register_token_validator(ZitadelIntrospectTokenValidator())
|
|
|
|
APP = Flask(__name__)
|
|
|
|
@APP.errorhandler(ValidatorError)
|
|
def handle_auth_error(ex: ValidatorError) -> Response:
|
|
|
|
response = jsonify(ex.error)
|
|
response.status_code = ex.status_code
|
|
return response
|
|
|
|
@APP.route("/img/api/public")
|
|
def public():
|
|
"""No access token required."""
|
|
response = (
|
|
"Public route - You don't need to be authenticated to see this."
|
|
)
|
|
return jsonify(message=response)
|
|
|
|
@APP.route("/img/api/private")
|
|
@require_auth(None)
|
|
def private():
|
|
"""A valid access token is required."""
|
|
response = (
|
|
"Private route - You need to be authenticated to see this."
|
|
)
|
|
return jsonify(message=response)
|
|
|
|
@APP.route("/img/api/private-scoped")
|
|
@require_auth(["read:messages"])
|
|
def private_scoped():
|
|
"""A valid access token and scope are required."""
|
|
response = (
|
|
"Private, scoped route - You need to be authenticated and have the role read:messages to see this."
|
|
)
|
|
return jsonify(message=response)
|
|
|
|
if __name__ == "__main__":
|
|
APP.run()
|
|
```
|
|
|
|
The API has three routes:
|
|
|
|
<ul>
|
|
<li> "/img/api/public" - No access token is required.</li>
|
|
<li>"/img/api/private" - A valid access token is required.</li>
|
|
<li>"/img/api/private-scoped" - A valid access token and a "read:messages" scope are required.</li>
|
|
</ul>
|
|
|
|
The [validator.py](https://github.com/zitadel/example-api-python3-flask/blob/main/validator.py) file implements the ZitadelIntrospectTokenValidator class, which is a custom class that inherits from the IntrospectTokenValidator class provided by the authlib library. The introspection process retrieves the token details from ZITADEL using ZITADEL's introspection endpoint.
|
|
|
|
```python
|
|
from os import environ as env
|
|
import os
|
|
import time
|
|
from typing import Dict
|
|
|
|
from authlib.oauth2.rfc7662 import IntrospectTokenValidator
|
|
import requests
|
|
from dotenv import load_dotenv, find_dotenv
|
|
from requests.auth import HTTPBasicAuth
|
|
|
|
load_dotenv()
|
|
|
|
ZITADEL_DOMAIN = os.getenv("ZITADEL_DOMAIN")
|
|
CLIENT_ID = os.getenv("CLIENT_ID")
|
|
CLIENT_SECRET = os.getenv("CLIENT_SECRET")
|
|
|
|
class ValidatorError(Exception):
|
|
|
|
def __init__(self, error: Dict[str, str], status_code: int):
|
|
super().__init__()
|
|
self.error = error
|
|
self.status_code = status_code
|
|
|
|
# Use Introspection in Resource Server
|
|
# https://docs.authlib.org/en/latest/specs/rfc7662.html#require-oauth-introspection
|
|
|
|
class ZitadelIntrospectTokenValidator(IntrospectTokenValidator):
|
|
def introspect_token(self, token_string):
|
|
url = f'{ZITADEL_DOMAIN}/oauth/v2/introspect'
|
|
data = {'token': token_string, 'token_type_hint': 'access_token', 'scope': 'openid'}
|
|
auth = HTTPBasicAuth(CLIENT_ID, CLIENT_SECRET)
|
|
resp = requests.post(url, data=data, auth=auth)
|
|
resp.raise_for_status()
|
|
return resp.json()
|
|
|
|
def match_token_scopes(self, token, or_scopes):
|
|
if or_scopes is None:
|
|
return True
|
|
roles = token["urn:zitadel:iam:org:project:roles"].keys()
|
|
for and_scopes in or_scopes:
|
|
scopes = and_scopes.split()
|
|
"""print(f"Check if all {scopes} are in {roles}")"""
|
|
if all(key in roles for key in scopes):
|
|
return True
|
|
return False
|
|
|
|
def validate_token(self, token, scopes, request):
|
|
print (f"Token: {token}\n")
|
|
now = int( time.time() )
|
|
if not token:
|
|
raise ValidatorError({
|
|
"code": "invalid_token_revoked",
|
|
"description": "Token was revoked." }, 401)
|
|
"""Expired"""
|
|
if token["exp"] < now:
|
|
raise ValidatorError({
|
|
"code": "invalid_token_expired",
|
|
"description": "Token has expired." }, 401)
|
|
"""Revoked"""
|
|
if not token["active"]:
|
|
raise InvalidTokenError()
|
|
"""Insufficient Scope"""
|
|
if not self.match_token_scopes(token, scopes):
|
|
raise ValidatorError({
|
|
"code": "insufficient_scope",
|
|
"description": f"Token has insufficient scope. Route requires: {scopes}" }, 401)
|
|
|
|
def __call__(self, *args, **kwargs):
|
|
res = self.introspect_token(*args, **kwargs)
|
|
return res
|
|
```
|
|
|
|
3. Create a new file named ".env" in the directory. Copy the settings in the [".env.example"](https://github.com/zitadel/example-api-python3-flask/blob/main/.env.example) file to the newly created .env file. Set the values with your Custom Domain/Issuer URL, Client ID, and Client Secret from the previous steps. Obtain your Issuer URL by following [these steps](/guides/start/quickstart#5-collect-your-integration-keys).
|
|
|
|
```python
|
|
ZITADEL_DOMAIN = "https://custom-domain-abcdef.zitadel.cloud"
|
|
CLIENT_ID = "197....@projectname"
|
|
CLIENT_SECRET = "NVAp70IqiGmJldbS...."
|
|
```
|
|
|
|
### ZITADEL settings to create a service account
|
|
|
|
|
|
|
|
1. Create a service account and a Personal Access Token (PAT) for that user by following [this guide](/guides/integrate/service-accounts/personal-access-token#create-a-service-account-with-a-pat).
|
|
2. To enable authorization, follow [this guide](/guides/manage/console/roles) to create a role `read:messages` on your project.
|
|
3. Next, add the role `read:messages` to the service account you created. Follow this [guide](/guides/manage/console/roles#role-assignments) for more information on creating a role assignment.
|
|
|
|
### Run the API
|
|
|
|
1. Install required dependencies by running `pip3 install -r requirements.txt` on your terminal.
|
|
2. Run the API with the `python3 server.py` command.
|
|
3. Open another terminal and follow the next step to test the API.
|
|
|
|
## Test the API
|
|
|
|
### Public route
|
|
|
|
Invoke the public route by running the following command:
|
|
|
|
```
|
|
curl --request GET \
|
|
--url http://127.0.0.1:5000/api/public
|
|
```
|
|
|
|
You should get a response with Status Code 200 and the following message.
|
|
|
|
`{"message":"Public route - You don't need to be authenticated to see this."}`
|
|
|
|
### Private route
|
|
|
|
Call the private route without authorization headers by running the following command:
|
|
|
|
```
|
|
curl --request GET \
|
|
--url http://127.0.0.1:5000/api/private
|
|
```
|
|
|
|
You should get a response with Status Code 401 and an error message.
|
|
|
|
Now let's add an authorization header to your request. Save the personal access token for your service account to a variable by running the following command. Replace the value with the PAT you obtained earlier.
|
|
|
|
`PAT=nr9vnUTkQkn4rxWk...`
|
|
|
|
Then call the private route with the PAT in the authorization header.
|
|
|
|
```
|
|
curl --request GET \
|
|
--url http://127.0.0.1:5000/api/private \
|
|
--header "authorization: Bearer $PAT"
|
|
```
|
|
|
|
Now you should get a response with Status Code 200 and the following message.
|
|
|
|
`{"message":"Private route - You need to be authenticated to see this."}`
|
|
|
|
### Private route, protected
|
|
|
|
Call the private route that requires the user to have a certain role
|
|
|
|
```
|
|
curl --request GET \
|
|
--url http://127.0.0.1:5000/api/private-scoped \
|
|
--header "authorization: Bearer $PAT"
|
|
```
|
|
|
|
You should get a response with Status Code 200 and the following message.
|
|
|
|
`{"message":"Private, scoped route - You need to be authenticated and have the role read:messages to see this."}`
|
|
|
|
You can remove the role from the service account in ZITADEL and try again. You should then get a Status Code 403, Forbidden error.
|