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': '[email protected]', '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

[seniuts-b2]

So, here’s a summary of the steps I took to make the setup work successfully. It can be valuable for someone else. But feel free to share with me better solution.

Env variables:

AAD_TENANT_ID = os.getenv("AAD_TENANT_ID")
AAD_CLIENT_ID = os.getenv("AAD_CLIENT_ID")
AAD_CLIENT_SECRET = os.getenv("AAD_CLIENT_SECRET")
# Airflow API Configuration
AIRFLOW_BASE_URL = "https://{{ your Airflow URL }}/auth/fab/v1"
# Azure AD OAuth2 Token URL
AUTH_URL = f"https://login.microsoftonline.com/{AAD_TENANT_ID}/oauth2/v2.0/token"
# Required permission scope for Airflow API
SCOPE = (
    f"api://{AAD_CLIENT_ID}/.default"
)
# Azure AD OAuth2 Token URL
TOKEN_URL = f"https://login.microsoftonline.com/{AAD_TENANT_ID}/oauth2/v2.0/token"

Here is my function to get Azure AD token:

import requests
import time

from oauthlib.oauth2 import BackendApplicationClient
from requests.exceptions import HTTPError, SSLError
from requests_oauthlib import OAuth2Session

# Function: Authenticate with Azure AD
# -----------------------------------------
def get_azure_ad_token_cached():
    """
    Authenticate with Azure AD using the client credentials flow and get an access token.

    Returns a valid Azure AD token. If a cached token exists and is still valid, returns the cached one. Otherwise, requests a new token from AAD.
    """
    token_cache = {"CACHED_TOKEN": None, "TOKEN_EXPIRES_AT": 0}

    # If there's a valid token in cache (and not expiring), return it
    if token_cache["CACHED_TOKEN"] and time.time() < token_cache["TOKEN_EXPIRES_AT"] - 60:
        return token_cache["CACHED_TOKEN"]

    # Otherwise, fetch a new token
    logger.info("Fetching a new token from Azure AD.")
    client = BackendApplicationClient(client_id=AAD_CLIENT_ID)
    oauth2_session = OAuth2Session(client=client)

    # payload = {
    #     "grant_type": "client_credentials",
    #     "client_id": AAD_CLIENT_ID,
    #     "client_secret": AAD_CLIENT_SECRET,
    #     "scope": SCOPE
    # }
    # logger.info(f"!!! JUST FOR TEST: Token request payload: {payload}")
    # try:
    #     response = requests.post(AUTH_URL, data=payload, timeout=10)
    #     logger.info(f"!!! JUST FOR TEST !!! Azure AD response:\n{response.text}")
    #     response.raise_for_status()
    #     token_response = response.json()
    #     return token_response["access_token"]
    # except HTTPError as http_err:
    #     logger.error(f"HTTP error while fetching token: {http_err}")
    #     raise http_err
    # except Exception as err:
    #     logger.error(f"Error obtaining Azure AD token: {err}")
    #     raise err

    # Fetch Access Token
    try:
        token_response = oauth2_session.fetch_token(
            token_url=TOKEN_URL,
            client_id=AAD_CLIENT_ID,
            client_secret=AAD_CLIENT_SECRET,
            scope=SCOPE,
        )
        logger.info("Token Acquired Successfully")
        # logger.info(f"!!! JUST FOR TEST !!! Azure AD token:\n{token_response}")
    except requests.exceptions.RequestException as e:
        logger.info(f"Failed to fetch token: {e}")
        raise requests.exceptions.RequestException(e) from e

    expires_in = token_response.get("expires_in", 3599)
    token_cache["TOKEN_EXPIRES_AT"] = time.time() + expires_in

    # Construct the actual bearer token string
    token_cache["CACHED_TOKEN"] = f"Bearer {token_response.get('access_token', '')}"

    return token_cache["CACHED_TOKEN"]

Here is the full code of API Auth backend:
Azure AD token overview

"""
Provides an authentication backend for Airflow using Azure Active Directory (Azure AD).

It includes functionality for validating Azure AD JWT tokens, fetching JWKS keys, and enforcing
authentication on API endpoints.
"""

from __future__ import annotations

import json
import logging
import os
from functools import wraps
from typing import Callable, TypeVar, cast

import jwt
import requests
from cryptography.hazmat.primitives import serialization
from flask import Response, current_app, request
from jwt.algorithms import RSAAlgorithm

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
# ------------------------------
# !!! IMPORTANT: AAD_TENANT_ID and AAD_CLIENT_ID parameters should be passed as environment variables through 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}/"  # 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"Fetching JWKS from: {JWKS_URL}")

# Cache JWKS keys to avoid frequent network requests
jwks_cache: dict[str, dict] = {}


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  # noqa: PLW0603
    if not jwks_cache:
        logger.info("Fetching JWKS from Azure AD...")
        response = requests.get(JWKS_URL, timeout=10)
        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.
    """
    logger.info("Validating Azure AD access token...")

    if len(token.split(".")) != 3:  # noqa: PLR2004
        logger.error("Invalid token format. Access denied.")
        return False

    # 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)
    # 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)
    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)
    public_key_bytes = public_key.public_bytes(
        encoding=serialization.Encoding.PEM, format=serialization.PublicFormat.SubjectPublicKeyInfo
    )

    try:
        ## Verify the expected jwt token claims and signature and return the decoded token claims: https://pyjwt.readthedocs.io/en/stable/api.html#jwt.decode
        decoded_token = jwt.decode(
            jwt=token,
            key=public_key_bytes,
            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": True,  # 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 decoded_token
    except jwt.ExpiredSignatureError as e:
        logger.error(f"Token has expired. Access denied: {e}")
    except jwt.InvalidAudienceError as e:
        logger.error(f"Invalid audience. Expected: {AUDIENCE}: {e}")
    except jwt.InvalidIssuerError as e:
        logger.error(f"Invalid issuer. Expected: {ISSUER}: {e}")
    except jwt.InvalidTokenError as e:
        logger.error(f"Invalid token error: {e}")
    return False


def lookup_airflow_user(token_claims: dict):
    """Finds an Airflow user or App ID based on token claims."""
    security_manager = current_app.appbuilder.sm  # Access the security manager

    # Get user from token (try email first, fallback to appid)
    user_email = token_claims.get("email") or token_claims.get("preferred_username")
    user_appid = token_claims.get("appid")  # App ID for service accounts
    logger.info(f"Looking for user with email: '{user_email}' or appid: '{user_appid}'")

    user = None
    if user_email:
        user = security_manager.find_user(email=user_email)
    if not user and user_appid:
        logger.info(f"Looking for service account user with appid: {user_appid}")
        user = security_manager.find_user(username=user_appid)  # Match appid to a user
    if not user:
        logger.error(f"User {user_email or user_appid} not found in Airflow. Access denied.")
        return None
    if not user.is_active:
        logger.error(f"User {user.username} is inactive. Access denied.")
        return None
    return user


def set_current_airflow_user(user):
    """Sets the current Airflow user in the request context."""
    current_app.appbuilder.sm.lm._update_request_context_with_user(user=user)


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")
        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]
        token_claims = validate_azure_token(token)
        if not token_claims:
            logger.error("Invalid or expired token. Access denied.")
            return Response("Unauthorized", 401, {"WWW-Authenticate": "Bearer"})

        user = lookup_airflow_user(token_claims)
        if not user:
            return Response("Forbidden", 403)

        set_current_airflow_user(user)

        logger.info("Authentication successful. Processing request.")
        return function(*args, **kwargs)

    return cast(T, decorated)


def init_app(_):
    """Initialize the authentication backend required by Airflow."""

Here are the needed updates in Azure App registrations :

• Add Application ID URI (Expose an API section):
  
• Add API permissions:
  

Permissions have to be requested for your API.
It means, you request/add Admin permissions thought App registration for the same App registrtions :

Admin is a group added in App roles :
)

Permissions have to be granted by the Admin (Admin consent required).
Adding Admin permissions to your API (App registrations) passes corresponded claim to AD token.

• Important Note: Changes made in App registrations (AppR) and assigning the Admin role to the AppR do not automatically grant access in Airflow. In other words, Airflow doesn’t recognize the AppR as a user by default.
  To fix this, we need to explicitly add the AppR's clientID as an Admin user in Airflow. You can do this by following these steps:
  • Connect to an Airflow pod in your Kubernetes cluster, e.g., the webserver:
    kubectl exec -it airflow-webserver-584df9db77-h65g5 -n airflow -- /bin/bash
  • Inside the pod, run the following Airflow CLI command:
    airflow users create --username clientID --password $(python3 -c 'import secrets; print(secrets.token_hex(16))') --firstname Service --lastname Account --role Admin --email [email protected]
    Here, --username should be set to the client ID of the AppR.
    We can generate own secure password as shown above, or simply use --use-random-password.
    Reference: Airflow CLI – create user

Once all of these were done, Airflow will recognize the AppR clientID as an Admin user, allowing it to authenticate and access Airflow APIs as expected.

[ChahatKumar]

where did you add the custom Airflow API Backend? I am trying to add as a plugin inn plugins folder or config map and both are not workking out for me