Authentication Backends

def init_app(app) -> None:
    """Initialize basic authentication for Flask application."""

def auth_current_user() -> User | None:
    """
    Authenticate current user using HTTP Basic Authentication.
    
    Returns:
        User object if authentication successful, None otherwise.
    """

def requires_authentication(function: T) -> T:
    """
    Decorator that requires basic authentication for the decorated function.
    
    Args:
        function: The function to be decorated.
        
    Returns:
        Decorated function that enforces basic authentication.
    """

class KerberosService:
    """Kerberos authentication service configuration."""
    
    def __init__(self): ...

class _KerberosAuth(NamedTuple):
    """Kerberos authentication result."""
    user: User
    token: str

def init_app(app) -> None:
    """Initialize Kerberos authentication for Flask application."""

def find_user(username: str | None = None, email: str | None = None) -> User | None:
    """
    Find user by username or email for Kerberos authentication.
    
    Args:
        username: Username to search for.
        email: Email address to search for.
        
    Returns:
        User object if found, None otherwise.
    """

def requires_authentication(
    function: T, 
    find_user: Callable[[str], BaseUser] | None = find_user
) -> T:
    """
    Decorator that requires Kerberos authentication for the decorated function.
    
    Args:
        function: The function to be decorated.
        find_user: Function to find user by username.
        
    Returns:
        Decorated function that enforces Kerberos authentication.
    """

def init_app(app) -> None:
    """Initialize session authentication for Flask application."""

def requires_authentication(function: T) -> T:
    """
    Decorator that requires session authentication for the decorated function.
    
    Args:
        function: The function to be decorated.
        
    Returns:
        Decorated function that enforces session authentication.
    """

from airflow.providers.fab.auth_manager.api.auth.backend import basic_auth
from flask import Flask

app = Flask(__name__)

# Initialize basic auth backend
basic_auth.init_app(app)

# Use the decorator on API endpoints
@basic_auth.requires_authentication
def get_dags():
    """API endpoint requiring basic authentication."""
    user = basic_auth.auth_current_user()
    if user:
        return {"message": f"Hello {user.username}"}
    return {"error": "Authentication failed"}, 401

from airflow.providers.fab.auth_manager.api.auth.backend import kerberos_auth
from flask import Flask

app = Flask(__name__)

# Initialize Kerberos auth backend
kerberos_auth.init_app(app)

# Use the decorator on API endpoints
@kerberos_auth.requires_authentication
def get_secure_data():
    """API endpoint requiring Kerberos authentication."""
    # User is available in Flask's g object after successful auth
    from flask import g
    return {"message": f"Authenticated as {g.user.username}"}

# Custom user lookup
def custom_find_user(username):
    """Custom user lookup function."""
    # Your custom logic to find user
    return kerberos_auth.find_user(username=username)

@kerberos_auth.requires_authentication(find_user=custom_find_user)
def custom_endpoint():
    """Endpoint with custom user lookup."""
    pass

from airflow.providers.fab.auth_manager.api.auth.backend import session
from flask import Flask

app = Flask(__name__)

# Initialize session auth backend
session.init_app(app)

# Use the decorator on web endpoints
@session.requires_authentication
def web_dashboard():
    """Web endpoint requiring session authentication."""
    # User is available through Flask-Login's current_user
    from flask_login import current_user
    return f"Welcome {current_user.get_full_name()}"

# In airflow.cfg or environment variables

[api]
# For basic auth
auth_backend = airflow.providers.fab.auth_manager.api.auth.backend.basic_auth

# For Kerberos auth  
auth_backend = airflow.providers.fab.auth_manager.api.auth.backend.kerberos_auth

# For session auth
auth_backend = airflow.providers.fab.auth_manager.api.auth.backend.session

from airflow.providers.fab.auth_manager.api.auth.backend.basic_auth import requires_authentication
from airflow.providers.fab.auth_manager.models import User

def custom_auth_current_user() -> User | None:
    """Custom authentication logic."""
    # Your custom authentication logic here
    # Return User object if authenticated, None otherwise
    pass

# Apply authentication decorator to your endpoints
@requires_authentication  
def my_api_endpoint():
    user = custom_auth_current_user()
    if user:
        return {"status": "authenticated", "user": user.username}
    return {"status": "failed"}, 401

from flask import jsonify

def handle_auth_error():
    """Handle authentication errors."""
    return jsonify({
        "error": "Authentication required",
        "message": "Please provide valid credentials"
    }), 401

def handle_forbidden():
    """Handle authorization errors."""
    return jsonify({
        "error": "Forbidden", 
        "message": "Insufficient permissions"
    }), 403

from typing import TypeVar, Callable, NamedTuple, TYPE_CHECKING
from airflow.auth.managers.models.base_user import BaseUser
from airflow.providers.fab.auth_manager.models import User

T = TypeVar("T", bound=Callable)

if TYPE_CHECKING:
    from flask import Flask

# Kerberos-specific configuration
KERBEROS_SERVICE_NAME = "HTTP"  # Service name for Kerberos
KERBEROS_HOSTNAME = None        # Hostname for Kerberos (None = auto-detect)

# No specific configuration required
# Uses standard HTTP Basic Authentication headers

# Uses Flask-Login session management
# Configured through Flask application settings
SECRET_KEY = "your-secret-key"  # Required for session encryption
SESSION_COOKIE_SECURE = True   # HTTPS only cookies
SESSION_COOKIE_HTTPONLY = True # No JavaScript access