Airflow API Returns 403 Forbidden When Using Azure AD Authentication via Custom API Backend

[seniuts-b2]

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

• I agree to follow this project's Code of Conduct

[boring-cyborg[bot]]

Thanks for opening your first issue here! Be sure to follow the issue template! If you are willing to raise PR to address this issue please do so, no need to wait for approval.

[potiuk]

that's an interesting discussion and maybe someone will be able to help, but I think expecting that somene will help with debugging your custom authentication is homestly I think a bit too much for people who are trying to help otehrs in their free time. Maybe somoene will help you to debug it, but unless you will distill it to something that will show that this is an airlfow and not your custom backend issue, you are mostly left to debugging your own code.

[potiuk]

Converted it to a discussion.

[seniut]

@potiuk thanks for your comment. I just cannot understand how I can see error if custom api backed finished successfully. I mean, I expect if I follow needed syntax to use custom api backed and instead of any validation just put ‘pass’ it should works …so,
I think the issue should be on the next step: maybe somewhere in Flask (FAB) or any thing else in Airflow.
But maybe I missed something and need more debugging, but I am not sure how to make this debug.
I will veer appreciate for any help.

[potiuk]

Someone's help posssibly, yes. I am not able to help you - don't have time to debug your code. But maybe somoene will, that's why it is converted to a discussion.

[koskelainen]

Hi @seniut
I have the same issue. I have a custom OAuth2 plugin that functions properly with the UI but not with the API.
My plugin: https://gist.github.com/koskelainen/57539ba7fba4c8e3e7234054872c26a4
I tested it for airflow version 2.9.1

airflow.cfg

...
[api]
enable_api = True
auth_backends = plugins.oauth2_custom.custom_security_manager, airflow.api.auth.backend.session, airflow.api.auth.backend.basic_auth
access_control_allow_headers = origin, Content-Type, accept, Authorization
access_control_allow_methods = POST, GET, OPTIONS, DELETE, PATCH
access_control_allow_origins = *
...

my error logs

[2025-02-26T10:51:38.209-0500] {flask_api.py:251} DEBUG - Getting data and status code
[2025-02-26T10:51:38.209-0500] {validation.py:168} DEBUG - https://dev.airflow.example.com:8090/api/v1/dags/example_dag1/dagRuns validating schema...
[2025-02-26T10:51:38.210-0500] {validation.py:365} DEBUG - https://dev.airflow.example.com:8090/api/v1/dags/example_dag1/dagRuns validating parameters...
[2025-02-26T10:51:38.210-0500] {parameter.py:85} DEBUG - Function Arguments: ['dag_id', 'session']
[2025-02-26T10:51:38.211-0500] {custom_security_manager.py:50} DEBUG - >>payload: {'aud': '**AAD_CLIENT_ID**', 'iss': 'https://login.microsoftonline.com/**AAD_TENANT_ID**/v2.0', 'iat': 1740581513, 'nbf': 1740581513, 'exp': 1740586150, 'aio': '***', 'azp': '***', 'azpacr': '0', 'name': 'User Name', 'oid': '***', 'preferred_username': 'user@example.com', 'rh': '***', 'roles': ['Admin'], 'scp': '***', 'sid': '***', 'sub': '***', 'tid': '***', 'uti': '***', 'ver': '2.0'}
[2025-02-26T10:51:38.214-0500] {abstract.py:280} DEBUG - Getting data and status code
[2025-02-26T10:51:38.214-0500] {abstract.py:423} DEBUG - Prepared body and status code (403)
[2025-02-26T10:51:38.214-0500] {abstract.py:292} DEBUG - Got framework response
10.101.1.123 - - [26/Feb/2025:10:51:38 -0500] "POST /api/v1/dags/example_dag1/dagRuns HTTP/1.1" 403 185 "-" "python-requests/2.32.3"

[point911]

Can reproduce the same issue, doesn't matter what I do, it's always 403, i created a new endpoint and made authorization via my custom requires_authentication and everything worked, however if I try to use it as a plugin - it doesn't work and return 403