0
# Auth Manager
1

2
The FabAuthManager is the core authentication manager that integrates Flask-AppBuilder's security framework with Airflow's authentication system. It provides user session management, authorization checks, and serves as the bridge between FAB's security model and Airflow's requirements.
3

4
## Capabilities
5

6
### Core Auth Manager
7

8
The main authentication manager class providing comprehensive user management and authorization.
9

10
```python { .api }
11
class FabAuthManager(BaseAuthManager):
12
    @staticmethod
13
    def get_cli_commands() -> list[CLICommand]:
14
        """Returns CLI commands to be included in Airflow CLI."""
15
        
16
    def get_api_endpoints(self) -> Blueprint | None:
17
        """Returns Flask Blueprint for API endpoints."""
18
        
19
    def get_user_display_name(self) -> str:
20
        """Returns the user's display name associated with the session."""
21
        
22
    def get_user(self) -> User:
23
        """Returns the user associated with the current session."""
24
        
25
    def init(self) -> None:
26
        """Initializes the auth manager."""
27
        
28
    def is_logged_in(self) -> bool:
29
        """Checks if the current user is authenticated."""
30
```
31

32
### Authorization Methods
33

34
Methods for checking user permissions across different Airflow resources.
35

36
```python { .api }
37
def is_authorized_configuration(
38
    self, 
39
    *, 
40
    method: ResourceMethod,
41
    details: ConfigurationDetails | None = None, 
42
    user: BaseUser | None = None
43
) -> bool:
44
    """Checks if user is authorized to access configuration."""
45

46
def is_authorized_connection(
47
    self, 
48
    *, 
49
    method: ResourceMethod,
50
    details: ConnectionDetails | None = None, 
51
    user: BaseUser | None = None
52
) -> bool:
53
    """Checks if user is authorized to access connection."""
54

55
def is_authorized_dag(
56
    self, 
57
    *, 
58
    method: ResourceMethod,
59
    access_entity: DagAccessEntity | None = None,
60
    details: DagDetails | None = None, 
61
    user: BaseUser | None = None
62
) -> bool:
63
    """Checks if user is authorized to access DAG."""
64

65
def is_authorized_asset(
66
    self, 
67
    *, 
68
    method: ResourceMethod,
69
    details: AssetDetails | None = None, 
70
    user: BaseUser | None = None
71
) -> bool:
72
    """Checks if user is authorized to access asset."""
73

74
def is_authorized_dataset(
75
    self, 
76
    *, 
77
    method: ResourceMethod,
78
    details: AssetDetails | None = None, 
79
    user: BaseUser | None = None
80
) -> bool:
81
    """Checks if user is authorized to access dataset.
82
    
83
    .. deprecated:: Airflow 3.0.0
84
        Use `is_authorized_asset` instead. This method will be removed in a future version.
85
    """
86

87
def is_authorized_pool(
88
    self, 
89
    *, 
90
    method: ResourceMethod,
91
    details: PoolDetails | None = None, 
92
    user: BaseUser | None = None
93
) -> bool:
94
    """Checks if user is authorized to access pool."""
95

96
def is_authorized_variable(
97
    self, 
98
    *, 
99
    method: ResourceMethod,
100
    details: VariableDetails | None = None, 
101
    user: BaseUser | None = None
102
) -> bool:
103
    """Checks if user is authorized to access variable."""
104

105
def is_authorized_view(
106
    self, 
107
    *, 
108
    access_view: AccessView, 
109
    user: BaseUser | None = None
110
) -> bool:
111
    """Checks if user is authorized to access view."""
112

113
def is_authorized_custom_view(
114
    self, 
115
    *, 
116
    method: ResourceMethod | str, 
117
    resource_name: str, 
118
    user: BaseUser | None = None
119
) -> bool:
120
    """Checks if user is authorized to access custom view."""
121
```
122

123
### Permission Management
124

125
Methods for retrieving and managing user permissions.
126

127
```python { .api }
128
def get_permitted_dag_ids(
129
    self, 
130
    *, 
131
    methods: Container[ResourceMethod] | None = None,
132
    user: BaseUser | None = None,
133
    session: Session = NEW_SESSION
134
) -> set[str]:
135
    """Returns set of DAG IDs that the user has permission to access."""
136
```
137

138
### Security Manager Integration
139

140
Access to the underlying FAB security manager for advanced operations.
141

142
```python { .api }
143
@cached_property
144
def security_manager(self) -> FabAirflowSecurityManagerOverride:
145
    """Returns the security manager instance."""
146
```
147

148
### URL Generation
149

150
Methods for generating authentication-related URLs.
151

152
```python { .api }
153
def get_url_login(self, **kwargs) -> str:
154
    """Returns the login URL with optional parameters."""
155

156
def get_url_logout(self) -> str:
157
    """Returns the logout URL."""
158
    
159
def register_views(self) -> None:
160
    """Registers views with the application."""
161

162
def get_url_user_profile(self) -> str | None:
163
    """Returns the user profile URL."""
164
```
165

166
## Usage Examples
167

168
### Basic Authentication Check
169

170
```python
171
from airflow.providers.fab.auth_manager.fab_auth_manager import FabAuthManager
172
from airflow.auth.managers.models.resource_details import DagDetails
173
from airflow.auth.managers.base_auth_manager import ResourceMethod
174

175
auth_manager = FabAuthManager()
176

177
# Check if user is logged in
178
if auth_manager.is_logged_in():
179
    user = auth_manager.get_user()
180
    print(f"User: {user.get_full_name()}")
181
    
182
    # Check DAG access
183
    dag_details = DagDetails(id="my_dag")
184
    if auth_manager.is_authorized_dag(method=ResourceMethod.GET, details=dag_details):
185
        print("User has access to my_dag")
186
```
187

188
### Permission Filtering
189

190
```python
191
# Get DAGs user has access to
192
permitted_dags = auth_manager.get_permitted_dag_ids(methods=[ResourceMethod.GET, ResourceMethod.PUT])
193
print(f"User can access: {permitted_dags}")
194
```
195

196
### URL Generation
197

198
```python
199
# Generate authentication URLs
200
login_url = auth_manager.get_url_login(next="/admin/")
201
logout_url = auth_manager.get_url_logout()
202
profile_url = auth_manager.get_url_user_profile()
203

204
print(f"Login: {login_url}")
205
print(f"Logout: {logout_url}")
206
print(f"Profile: {profile_url}")
207
```
208

209
## Types
210

211
```python { .api }
212
from typing import TYPE_CHECKING, Container
213
from functools import cached_property
214
from airflow.auth.managers.models.base_user import BaseUser
215
from airflow.auth.managers.base_auth_manager import ResourceMethod
216
from airflow.auth.managers.models.resource_details import (
217
    DagDetails, AccessView, ConnectionDetails, VariableDetails, 
218
    PoolDetails, ConfigurationDetails, DagAccessEntity
219
)
220
from airflow.cli.cli_config import CLICommand
221
from flask import Blueprint
222
from sqlalchemy.orm import Session
223
from airflow.utils.session import NEW_SESSION
224

225
if TYPE_CHECKING:
226
    from airflow.providers.fab.auth_manager.security_manager.override import FabAirflowSecurityManagerOverride
227
    from airflow.providers.common.compat.assets import AssetDetails
228
```