You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "seniuts-b2 (via GitHub)" <gi...@apache.org> on 2025/02/24 17:53:10 UTC
Re: [D] Airflow API Returns 403 Forbidden When Using Azure AD Authentication via Custom API Backend [airflow]
GitHub user seniuts-b2 edited a discussion: Airflow API Returns 403 Forbidden When Using Azure AD Authentication via Custom API Backend
### Apache Airflow version
2.10.5
### If "Other Airflow 2 version" selected, which one?
2.10.4
### What happened?
I deployed Airflow on Azure K8s (AKS) via the Airflow Official Helm Chart. For UI authentication I use Azure AD via OAuth2 for that I have Azure App Registration for handling AIrflow access via Role-based access control (RBAC). Everything works as expected.
There are a couple of settings in Azure App Registry:
API permissions:
there are permissions for access to UI through `Delegated` permission type and `Application` permission type to Airflow API access via custom API backend through Application URI ID (using as a `scope` to fetch access token):
So, I use a custom Airflow API backend to access Airflow and manage Roles via endpoint:` https://airflow-ui-test.westeurope.cloudapp.azure.com/auth/fab/v1/roles`. In values.yaml Helm chart file in env section I have:
```
- name: AIRFLOW__API__AUTH_BACKENDS
value: "airflow_utility.dag_level_access_control.azure_ad_auth_backend,airflow.api.auth.backend.session"
```
here is my custom API backend: `airflow_utility.dag_level_access_control.azure_ad_auth_backend`
And I get error:
```
{
"detail": null,
"status": 403,
"title": "Forbidden",
"type": "https://airflow.apache.org/docs/apache-airflow/2.10.5/stable-rest-api-ref.html#section/Errors/PermissionDenied"
}
```
There is the custom Airflow API Backend:
```
from __future__ import annotations
import os
import json
import requests
import jwt
import logging
from jwt.algorithms import RSAAlgorithm
from flask import Response, request
from functools import wraps
from typing import TYPE_CHECKING, Any, Callable, TypeVar, cast
T = TypeVar("T", bound=Callable) # TypeVar for function decorators
# ------------------------------
# Logging Configuration
# ------------------------------
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logger = logging.getLogger(__name__)
# ------------------------------
# Azure AD Configuration
# ------------------------------
# AAD_TENANT_ID and AAD_CLIENT_ID parameters passed as environment variables though the values.yaml file webserver section
AAD_TENANT_ID = os.getenv("AAD_TENANT_ID")
AAD_CLIENT_ID = os.getenv("AAD_CLIENT_ID")
JWKS_URL = f"https://login.microsoftonline.com/{AAD_TENANT_ID}/discovery/v2.0/keys" # JSON Web Key Set (JWKS) URL
ISSUER = f"https://sts.windows.net/{AAD_TENANT_ID}/" #f"https://login.microsoftonline.com/{AAD_TENANT_ID}/v2.0" # Issuer (iss) claim in the token
AUDIENCE = f"api://{AAD_CLIENT_ID}" # Expected audience (should match the Application ID URI)
LEEWAY_SECONDS = 60 # Allow 60 seconds of clock drift for `exp`, `nbf`, `iat` validation
if not AAD_TENANT_ID or not AAD_CLIENT_ID:
msg = "Missing required environment variables (should be passed as env variable through values.yaml file webserver secrion): AAD_TENANT_ID, AAD_CLIENT_ID"
logger.error(msg)
raise ValueError(msg)
# logger.info(f"Using AAD_TENANT_ID: {AAD_TENANT_ID}")
# logger.info(f"Using CLIENT_ID: {AAD_CLIENT_ID}")
logger.info(f"Fetching JWKS from: {JWKS_URL}")
# Cache JWKS keys to avoid frequent network requests
jwks_cache = {}
def get_azure_jwks():
"""
Fetch and cache Azure AD JWKS (JSON Web Key Set) to verify token signatures.
JWKS is a public key repository published by Azure AD.
It contains multiple keys that are used to verify JWT signatures.
Azure AD JWKS URL looks like: https://login.microsoftonline.com/{AAD_TENANT_ID}/discovery/v2.0/keys
Azure AD rotates keys from time to time, so fetching them once per process is enough:
https://learn.microsoft.com/en-us/entra/identity-platform/signing-key-rollover#:~:text=metadata%20document.-,For%20security%20purposes%2C%20the%20Microsoft%20identity%20platform%E2%80%99s%20signing%20key%20rolls%20on,a%20key%20rollover%20event%20no%20matter%20how%20frequently%20it%20may%20occur.,-If%20your%20application
"""
global jwks_cache
if not jwks_cache:
logger.info("Fetching JWKS from Azure AD...")
response = requests.get(JWKS_URL)
response.raise_for_status()
jwks_cache = response.json()
logger.info("JWKS fetched and cached.")
return jwks_cache
def validate_azure_token(token: str) -> bool:
"""
token (str): there is received Airflow API request token of Azure AD JWT access token.
JWT (JSON Web Token) has three parts:
# Header (contains metadata like algorithm and key ID kid)
# Payload (contains claims like iss, exp, aud, etc.)
# Signature (ensures integrity)
Validate an Azure AD JWT access token workflow:
1. Extract the JWT Header → Get the "kid"
2. Fetch JWKS from Azure → Get the list of public keys
3. Find the matching kid → Use it to verify the JWT
4. Decode JWT using the correct key → Validate signature and claims
5. If everything is valid, grant access
Short workflow: The function fetches the JWKS, finds the correct key, and verifies the token signature.
"""
try:
logger.info("Validating Azure AD access token...")
# Decode the JWT header to get the key ID (kid):
## The JWT Header contains a "kid" (Key ID) field, which tells which public key was used to sign the JWT.
header = jwt.get_unverified_header(token)
logger.info(f"Token: {token}")
logger.info(f"Token header: {header}")
logger.info(f"Token key ID (kid): {header['kid']}")
# Fetch JWKS and find the correct key.
jwks = get_azure_jwks()
# logger.info(f"JWKS: {jwks}")
key = next((json.dumps(k) for k in jwks["keys"] if k["kid"] == header["kid"]), None)
logger.info(f"JWKS key: {key}")
if not key:
logger.error("Token key ID (kid) not found in JWKS. Token is invalid.")
return False
logger.info("Matching JWKS key found. Verifying token signature...")
# Decode and verify the token:
public_key = RSAAlgorithm.from_jwk(key)
## Verify the jwt token signature and return the token claims: https://pyjwt.readthedocs.io/en/stable/api.html#jwt.decode
decoded_token = jwt.decode(
jwt=token,
key=public_key,
algorithms=["RS256"],
# extended decoding and validation options
audience=AUDIENCE, # Ensures token is meant for this API
issuer=ISSUER, # Ensures token was issued by Azure AD
# leeway=LEEWAY_SECONDS, # Allow 60 seconds of clock drift for exp/nbf validation
options={
"verify_signature": False, # Ensure signature is verified
"require": ["exp", "iat", "nbf"], # Ensure these claims exist
"verify_exp": True, # Ensure token is not expired
"verify_iat": True, # Ensure issued-at (iat) is valid
"verify_nbf": True, # Ensure not-before (nbf) is valid
"verify_aud": True, # Ensure audience (aud) claim is valid
"verify_iss": True, # Ensure issuer (iss) claim is valid
"strict_aud": False # check that the aud claim is a single value (not a list), and matches audience exactly
}
)
logger.info(f"Decoded token: {decoded_token}")
logger.info("Token is valid and contains required scope. Access granted.")
return True # Token is valid
except jwt.ExpiredSignatureError as e:
logger.error(f"Token has expired. Access denied: {e}")
return False
except jwt.InvalidAudienceError as e:
logger.error(f"Invalid audience. Expected: {AUDIENCE}: {e}")
return False
except jwt.InvalidIssuerError as e:
logger.error(f"Invalid issuer. Expected: {ISSUER}: {e}")
return False
except jwt.InvalidTokenError as e:
logger.error(f"Invalid token error: {e}")
return False
def requires_authentication(function: T):
"""
Decorator to enforce authentication on API endpoints using Azure AD token.
"""
@wraps(function)
def decorated(*args, **kwargs):
logger.info(" ### Custome Airflow API Backend to validate Azure AD Authentication ###")
logger.info(f"Executing function: {function.__name__}") # Extract function name
logger.info(f"request: {request}")
auth_header = request.headers.get("Authorization")
logger.info(f"auth_header: {auth_header}")
if not auth_header or not auth_header.startswith("Bearer "):
logger.error("Missing or malformed Authorization header. Access denied.")
return Response("Unauthorized", 401, {"WWW-Authenticate": "Bearer"})
# As the token is the part of Authorization header we can extract it by splitting the header (authorization header is in the format "Bearer <token>")
token = auth_header.split(" ")[1]
if not validate_azure_token(token):
logger.error("Invalid or expired token. Access denied.")
return Response("Unauthorized", 401, {"WWW-Authenticate": "Bearer"})
logger.info("Authentication successful. Processing request.")
return function(*args, **kwargs)
return cast(T, decorated)
# return decorated
def init_app(_):
"""
Initialize the authentication backend (required by Airflow).
"""
```
I use `"verify_signature": False` because using `jwt.decode` function I get an every time error that the signature is invalid. If someone knows why and how to fix it, I would be glad to get any recommendations. But this isn't the main question.
The main question is:
Why do I get the error above? I expect to have access if I use a custom API backend to validate the token. The gotten token includes claim `roles` (for mapping Airflow roles and App registration roles) as I use `Application` permission type in App registry API permissions.
There are a couple of logs from API backend:
```
[2025-02-24T13:00:04.015+0000] {azure_ad_auth_backend.py:153} INFO - ### Custome Airflow API Backend to validate Azure AD Authentication ###
[2025-02-24T13:00:04.015+0000] {azure_ad_auth_backend.py:154} INFO - Executing function: Response
[2025-02-24T13:00:04.015+0000] {azure_ad_auth_backend.py:155} INFO - request: <Request 'https://airflow-ui-test.westeurope.cloudapp.azure.com/api/v1/roles/DataEngineer' [GET]>
[2025-02-24T13:00:04.015+0000] {azure_ad_auth_backend.py:158} INFO - auth_header: Bearer ...{{token}}...
[2025-02-24T13:00:04.016+0000] {azure_ad_auth_backend.py:85} INFO - Validating Azure AD access token...
[2025-02-24T13:00:04.016+0000] {azure_ad_auth_backend.py:90} INFO - Token: ...{token}...
[2025-02-24T13:00:04.016+0000] {azure_ad_auth_backend.py:91} INFO - Token header: {'typ': 'JWT', 'alg': 'RS256', 'x5t': 'imi0Y2z0dYKxBttAqK_Tt5hYBTk', 'kid': '{{ some kid}}'}
[2025-02-24T13:00:04.016+0000] {azure_ad_auth_backend.py:92} INFO - Token key ID (kid): {{ some kid}}
[2025-02-24T13:00:04.016+0000] {azure_ad_auth_backend.py:98} INFO - JWKS key: {"kty": "RSA", "use": "sig", "kid": "{{some kid}}", "x5t": "{{some values}}", "n": "{{some values}}", "e": "AQAB", "x5c": ["{{some values}}"], "cloud_instance_name": "microsoftonline.com", "issuer": "https://login.microsoftonline.com/{{ tenant_id }}/v2.0"}
[2025-02-24T13:00:04.017+0000] {azure_ad_auth_backend.py:102} INFO - Matching JWKS key found. Verifying token signature...
[2025-02-24T13:00:04.017+0000] {azure_ad_auth_backend.py:107} INFO - !!!! Public key: <cryptography.hazmat.bindings._rust.openssl.rsa.RSAPublicKey object at 0x7f11e134aa50>
[2025-02-24T13:00:04.017+0000] {azure_ad_auth_backend.py:128} INFO - Decoded token: {'aud': 'api://{{ app client id }}', 'iss': 'https://sts.windows.net/{{ tenant_id }}/', 'iat': 1740401432, 'nbf': 1740401432, 'exp': 1740405332, 'aio': '{{some values}}', 'appid': '{{ app client id }}', 'appidacr': '1', 'idp': 'https://sts.windows.net/{{ tenant id }}/', 'oid': '{{some values}}', 'rh': '{{some values}}', 'roles': ['Admin'], 'sub': '{{some values}}', 'tid': '{{some values}}', 'uti': '{{some values}}', 'ver': '1.0'}
[2025-02-24T13:00:04.017+0000] {azure_ad_auth_backend.py:130} INFO - Token is valid and contains required scope. Access granted.
[2025-02-24T13:00:04.018+0000] {azure_ad_auth_backend.py:171} INFO - Authentication successful. Processing request.
10.224.0.199 - - [24/Feb/2025:13:00:04 +0000] "GET /api/v1/roles/DataEngineer HTTP/1.1" 403 186 "-" "PostmanRuntime/7.43.0"
```
I get the same error:
```
{
"detail": null,
"status": 403,
"title": "Forbidden",
"type": "https://airflow.apache.org/docs/apache-airflow/2.10.5/stable-rest-api-ref.html#section/Errors/PermissionDenied"
}
```
using Postman and just pure python through GET request and {"Authorization": f"Bearer {access_token}"} in header
### What you think should happen instead?
I expect that if I use the Airflow API backend I should have access to API if the backend code was passed.``
### How to reproduce
deploy Airflow on AKS and use custom API backend to use Azure AD token.
### Operating System
PRETTY_NAME="Debian GNU/Linux 12 (bookworm)" NAME="Debian GNU/Linux" VERSION_ID="12" VERSION="12 (bookworm)" VERSION_CODENAME=bookworm ID=debian HOME_URL="https://www.debian.org/" SUPPORT_URL="https://www.debian.org/support" BUG_REPORT_URL="https://bugs.debian.org/"
### Versions of Apache Airflow Providers
airflow@airflow-webserver-7fdffbcd6b-cznnc:/opt/airflow$ pip freeze | grep apache-airflow-provider
apache-airflow-providers-amazon==9.2.0
apache-airflow-providers-celery==3.10.0
apache-airflow-providers-cncf-kubernetes==10.1.0
apache-airflow-providers-common-compat==1.3.0
apache-airflow-providers-common-io==1.5.0
apache-airflow-providers-common-sql==1.21.0
apache-airflow-providers-docker==4.0.0
apache-airflow-providers-elasticsearch==6.0.0
apache-airflow-providers-fab==1.5.2
apache-airflow-providers-ftp==3.12.0
apache-airflow-providers-google==12.0.0
apache-airflow-providers-grpc==3.7.0
apache-airflow-providers-hashicorp==4.0.0
apache-airflow-providers-http==5.0.0
apache-airflow-providers-imap==3.8.0
apache-airflow-providers-microsoft-azure==10.3.0
apache-airflow-providers-mysql==6.0.0
apache-airflow-providers-odbc==4.9.0
apache-airflow-providers-openlineage==2.0.0
apache-airflow-providers-postgres==6.0.0
apache-airflow-providers-redis==4.0.0
apache-airflow-providers-sendgrid==4.0.0
apache-airflow-providers-sftp==5.0.0
apache-airflow-providers-slack==9.0.0
apache-airflow-providers-smtp==1.9.0
apache-airflow-providers-snowflake==6.0.0
apache-airflow-providers-sqlite==4.0.0
apache-airflow-providers-ssh==4.0.0
### Deployment
Official Apache Airflow Helm Chart
### Deployment details
AKS, Official Airflow Helm Chart.
### Anything else?
_No response_
### Are you willing to submit PR?
- [ ] Yes I am willing to submit a PR!
### Code of Conduct
- [x] I agree to follow this project's [Code of Conduct](https://github.com/apache/airflow/blob/main/CODE_OF_CONDUCT.md)
GitHub link: https://github.com/apache/airflow/discussions/47029
----
This is an automatically sent email for commits@airflow.apache.org.
To unsubscribe, please send an email to: commits-unsubscribe@airflow.apache.org