airflow_fab_auth_manager_rbac_permission_check_quickstart.py

1import os
2from airflow.models import DagBag
3from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager
4from airflow.providers.fab.auth_manager.security_manager.override import FabAirflowSecurityManagerOverride
5
6# Note: In a real environment, ensure your airflow.cfg contains:
7# [core]
8# auth_manager = airflow.providers.fab.auth_manager.fab_auth_manager.FabAuthManager
9
10def quickstart_fab_example():
11    # 1. Initialize the FAB Auth Manager
12    # In Airflow 2.9+, FAB is the default provider for UI and Auth
13    auth_manager = FabAuthManager(None)
14    
15    # 2. Access the Security Manager
16    # This is the core component of the FAB provider used for RBAC
17    security_manager = auth_manager.security_manager
18    
19    # 3. Example: Programmatically check if a user has access to a DAG
20    # This mimics the internal logic used by the Airflow UI
21    dag_id = "example_bash_operator"
22    user_name = "admin"
23    
24    print(f"Checking permissions for user: {user_name} on DAG: {dag_id}")
25    
26    # In a real scenario, you would fetch a User object from the DB
27    # user = security_manager.find_user(username=user_name)
28    
29    # Example check for 'can_read' permission on a specific DAG resource
30    has_access = security_manager.has_access("can_read", "DAGs", user_name)
31    
32    if has_access:
33        print(f"User {user_name} is authorized to view {dag_id}.")
34    else:
35        print(f"User {user_name} is NOT authorized.")
36
37if __name__ == "__main__":
38    # Ensure Airflow home is set for initialization
39    if "AIRFLOW_HOME" not in os.environ:
40        os.environ["AIRFLOW_HOME"] = "~/airflow"
41        
42    try:
43        quickstart_fab_example()
44    except Exception as e:
45        print(f"Quickstart requires an initialized Airflow DB: {e}")