I’ve created a custom rest api endpoint in airflow. Its working perfect. I’ve authenticated it using the decorator ‘endpoint_bp.before_request’.

I’ve hard coded the user credentials in ‘check_auth’ function.

from airflow.plugins_manager import AirflowPlugin
from airflow.www.app import csrf
from flask import Blueprint, jsonify, request

# Define a Flask blueprint for your custom endpoint
endpoint_bp = Blueprint("my_plugin", __name__)


def check_auth(username, password):
    return username == "<username>" and password == "<password>"


@endpoint_bp.before_request
def require_basic_auth():
    auth = request.authorization
    if not auth or not check_auth(auth.username, auth.password):
        return jsonify({"message": "UnAuthorized"}), 401


@endpoint_bp.route("/api/v1/custom-end-point", methods=["POST"])
@csrf.exempt
def custom_endpoint():
    data, status = {}, 200
    return jsonify(data), status

# Define a custom plugin class to register the blueprint
class CustomPlugin(AirflowPlugin):
    """custom plugin class"""

    name = "custom_plugin"
    flask_blueprints = [endpoint_bp]

How can I use airflow’s basic authentication/web authentication to fix it.

Is there any way to use airflow metadatabase and check for the user credentials either using existing methods or in some other way,
I tried getting it from the database, but I’m not sure how to compare the encrypted password in db with the user provided(I’ve only one user in db as of now). One way I think about encrypt the password string and compare it with the one in db.
How can I do it?

Is existing method in airflow ( which is using in airflow web).
More precisely I’m looking for something similar in django

from django.contrib.auth import authenticate
user = authenticate(username="john", password="secret")
if user is not None:
    print("success")