Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/auth/managers/base_auth_manager.py: 81%
218 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-07 14:22 +0000
1#
2# Licensed to the Apache Software Foundation (ASF) under one
3# or more contributor license agreements. See the NOTICE file
4# distributed with this work for additional information
5# regarding copyright ownership. The ASF licenses this file
6# to you under the Apache License, Version 2.0 (the
7# "License"); you may not use this file except in compliance
8# with the License. You may obtain a copy of the License at
9#
10# http://www.apache.org/licenses/LICENSE-2.0
11#
12# Unless required by applicable law or agreed to in writing,
13# software distributed under the License is distributed on an
14# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15# KIND, either express or implied. See the License for the
16# specific language governing permissions and limitations
17# under the License.
18from __future__ import annotations
20import logging
21import warnings
22from abc import ABCMeta, abstractmethod
23from collections import defaultdict
24from enum import Enum
25from functools import cache
26from typing import TYPE_CHECKING, Any, Generic, Literal, TypeVar
28from jwt import InvalidTokenError
29from sqlalchemy import select
31from airflow.api_fastapi.auth.managers.models.base_user import BaseUser
32from airflow.api_fastapi.auth.managers.models.resource_details import (
33 ConnectionDetails,
34 DagDetails,
35 PoolDetails,
36 TeamDetails,
37 VariableDetails,
38)
39from airflow.api_fastapi.auth.tokens import (
40 JWTGenerator,
41 JWTValidator,
42 get_sig_validation_args,
43 get_signing_args,
44)
45from airflow.api_fastapi.common.types import ExtraMenuItem, MenuItem
46from airflow.configuration import conf
47from airflow.models import Connection, DagModel, Pool, Variable
48from airflow.models.dagbundle import DagBundleModel
49from airflow.models.revoked_token import RevokedToken
50from airflow.models.team import Team, dag_bundle_team_association_table
51from airflow.typing_compat import Unpack
52from airflow.utils.log.logging_mixin import LoggingMixin
53from airflow.utils.session import NEW_SESSION, provide_session
55if TYPE_CHECKING: 55 ↛ 56line 55 didn't jump to line 56 because the condition on line 55 was never true
56 from collections.abc import Sequence
58 from fastapi import FastAPI
59 from sqlalchemy import Row
60 from sqlalchemy.orm import Session
61 from starlette.middleware import _MiddlewareFactory
63 from airflow.api_fastapi.auth.managers.models.batch_apis import (
64 IsAuthorizedConnectionRequest,
65 IsAuthorizedDagRequest,
66 IsAuthorizedPoolRequest,
67 IsAuthorizedVariableRequest,
68 )
69 from airflow.api_fastapi.auth.managers.models.resource_details import (
70 AccessView,
71 AssetAliasDetails,
72 AssetDetails,
73 ConfigurationDetails,
74 DagAccessEntity,
75 )
76 from airflow.cli.cli_config import CLICommand
78if TYPE_CHECKING: 78 ↛ 80line 78 didn't jump to line 80 because the condition on line 78 was never true
79 # For static type checking - accepts string literals
80 ResourceMethod = Literal["GET", "POST", "PUT", "DELETE"]
81 ExtendedResourceMethod = Literal["GET", "POST", "PUT", "DELETE", "MENU"]
82else:
83 # For runtime - provides iteration and validation
85 class ResourceMethod(str, Enum):
86 """HTTP methods (actions) a user can perform against a resource."""
88 GET = "GET"
89 POST = "POST"
90 PUT = "PUT"
91 DELETE = "DELETE"
93 def __str__(self) -> str:
94 return self.value
96 class ExtendedResourceMethod(str, Enum):
97 """Extended HTTP methods including MENU for UI resource authorization."""
99 GET = "GET"
100 POST = "POST"
101 PUT = "PUT"
102 DELETE = "DELETE"
103 MENU = "MENU"
105 def __str__(self) -> str:
106 return self.value
109log = logging.getLogger(__name__)
110T = TypeVar("T", bound=BaseUser)
113COOKIE_NAME_JWT_TOKEN = "_token"
116class BaseAuthManager(Generic[T], LoggingMixin, metaclass=ABCMeta):
117 """
118 Class to derive in order to implement concrete auth managers.
120 Auth managers are responsible for any user management related operation such as login, logout, authz, ...
121 """
123 def init(self) -> None:
124 """Run operations when Airflow is initializing."""
125 if conf.getboolean("core", "multi_team"): 125 ↛ 126line 125 didn't jump to line 126 because the condition on line 125 was never true
126 am_teams = self._get_teams()
127 db_teams = Team.get_all_team_names()
129 if not db_teams.issuperset(am_teams):
130 raise ValueError(
131 f"Teams defined in the auth manager ({am_teams}) are not present in the database ({db_teams})."
132 )
134 @abstractmethod
135 def deserialize_user(self, token: dict[str, Any]) -> T:
136 """Create a user object from dict."""
138 @abstractmethod
139 def serialize_user(self, user: T) -> dict[str, Any]:
140 """Create a subject and extra claims dict from a user object."""
142 def revoke_token(self, token: str) -> None:
143 """Revoke a JWT token by persisting its JTI in the database."""
144 self._get_token_validator().revoke_token(token)
146 async def get_user_from_token(self, token: str) -> BaseUser:
147 """Verify the JWT token is valid and create a user object from it if valid."""
148 try:
149 payload: dict[str, Any] = await self._get_token_validator().avalidated_claims(token)
150 except InvalidTokenError as e:
151 log.error("JWT token is not valid: %s", e)
152 raise e
154 if (jti := payload.get("jti")) and RevokedToken.is_revoked(jti): 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true
155 raise InvalidTokenError("Token has been revoked")
157 try:
158 return self.deserialize_user(payload)
159 except (ValueError, KeyError) as e:
160 log.error("Couldn't deserialize user from token, JWT token is not valid: %s", e)
161 raise InvalidTokenError(str(e))
163 def get_fastapi_middlewares(self) -> list[tuple[_MiddlewareFactory[Any], dict[str, Any]]]:
164 """
165 Return middlewares the auth manager wants registered on the main FastAPI app.
167 Each entry is a ``(middleware_class, kwargs)`` tuple and is registered via
168 ``app.add_middleware`` by the API server. Auth managers that need to intercept or
169 augment incoming requests (for example, attaching an anonymous user to
170 unauthenticated requests when public access is configured) should override this
171 method.
172 """
173 return []
175 def generate_jwt(
176 self, user: T, *, expiration_time_in_seconds: int = conf.getint("api_auth", "jwt_expiration_time")
177 ) -> str:
178 """Return the JWT token from a user object."""
179 return self._get_token_signer(expiration_time_in_seconds=expiration_time_in_seconds).generate(
180 self.serialize_user(user)
181 )
183 @abstractmethod
184 def get_url_login(self, **kwargs) -> str:
185 """Return the login page url."""
187 def get_url_logout(self) -> str | None:
188 """
189 Return the logout page url.
191 The user is redirected to this URL when logging out. If None is returned (by default), no redirection
192 is performed. This redirection is usually needed to invalidate resources when logging out, such as a
193 session.
194 """
195 return None
197 def refresh_user(self, *, user: T) -> T | None:
198 """
199 Refresh the user if needed.
201 By default, does nothing. Some auth managers might need to refresh the user to, for instance,
202 refresh some tokens that are needed to communicate with a service/tool.
204 This method is called by every single request, it must be lightweight otherwise the overall API
205 server latency will increase.
206 """
207 return None
209 @abstractmethod
210 def is_authorized_configuration(
211 self,
212 *,
213 method: ResourceMethod,
214 user: T,
215 details: ConfigurationDetails | None = None,
216 ) -> bool:
217 """
218 Return whether the user is authorized to perform a given action on configuration.
220 :param method: the method to perform
221 :param user: the user to performing the action
222 :param details: optional details about the configuration
223 """
225 @abstractmethod
226 def is_authorized_connection(
227 self,
228 *,
229 method: ResourceMethod,
230 user: T,
231 details: ConnectionDetails | None = None,
232 ) -> bool:
233 """
234 Return whether the user is authorized to perform a given action on a connection.
236 :param method: the method to perform
237 :param user: the user to performing the action
238 :param details: optional details about the connection
239 """
241 @abstractmethod
242 def is_authorized_dag(
243 self,
244 *,
245 method: ResourceMethod,
246 user: T,
247 access_entity: DagAccessEntity | None = None,
248 details: DagDetails | None = None,
249 ) -> bool:
250 """
251 Return whether the user is authorized to perform a given action on a Dag.
253 :param method: the method to perform
254 :param user: the user to performing the action
255 :param access_entity: the kind of Dag information the authorization request is about.
256 If not provided, the authorization request is about the Dag itself
257 :param details: optional details about the Dag
258 """
260 @abstractmethod
261 def is_authorized_asset(
262 self,
263 *,
264 method: ResourceMethod,
265 user: T,
266 details: AssetDetails | None = None,
267 ) -> bool:
268 """
269 Return whether the user is authorized to perform a given action on an asset.
271 :param method: the method to perform
272 :param user: the user to performing the action
273 :param details: optional details about the asset
274 """
276 @abstractmethod
277 def is_authorized_asset_alias(
278 self,
279 *,
280 method: ResourceMethod,
281 user: T,
282 details: AssetAliasDetails | None = None,
283 ) -> bool:
284 """
285 Return whether the user is authorized to perform a given action on an asset alias.
287 :param method: the method to perform
288 :param user: the user to perform the action on
289 :param details: optional details about the asset alias
290 """
292 @abstractmethod
293 def is_authorized_pool(
294 self,
295 *,
296 method: ResourceMethod,
297 user: T,
298 details: PoolDetails | None = None,
299 ) -> bool:
300 """
301 Return whether the user is authorized to perform a given action on a pool.
303 :param method: the method to perform
304 :param user: the user to performing the action
305 :param details: optional details about the pool
306 """
308 def is_authorized_team(
309 self,
310 *,
311 method: ResourceMethod,
312 user: T,
313 details: TeamDetails | None = None,
314 ) -> bool:
315 """
316 Return whether the user is authorized to perform a given action on a team.
318 It is used primarily to check whether a user belongs to a team.
319 This function needs to be overridden by an auth manager compatible with multi-team.
321 :param method: the method to perform
322 :param user: the user performing the action
323 :param details: optional details about the team
324 """
325 raise NotImplementedError(
326 "The auth manager you are using is not compatible with multi-team. "
327 "In order to run Airflow in multi-team mode you need to use an auth manager compatible with it."
328 )
330 @abstractmethod
331 def is_authorized_variable(
332 self,
333 *,
334 method: ResourceMethod,
335 user: T,
336 details: VariableDetails | None = None,
337 ) -> bool:
338 """
339 Return whether the user is authorized to perform a given action on a variable.
341 :param method: the method to perform
342 :param user: the user to performing the action
343 :param details: optional details about the variable
344 """
346 @abstractmethod
347 def is_authorized_view(
348 self,
349 *,
350 access_view: AccessView,
351 user: T,
352 ) -> bool:
353 """
354 Return whether the user is authorized to access a read-only state of the installation.
356 :param access_view: the specific read-only view/state the authorization request is about.
357 :param user: the user to performing the action
358 """
360 @abstractmethod
361 def is_authorized_custom_view(self, *, method: ResourceMethod, resource_name: str, user: T) -> bool:
362 """
363 Return whether the user is authorized to perform a given action on a custom view.
365 A custom view can be a view defined as part of the auth manager. This view is then only available when
366 the auth manager is used as part of the environment. It can also be a view defined as part of a
367 plugin defined by a user.
369 :param method: the method to perform.
370 The method can also be a string if the action has been defined in a plugin.
371 In that case, the action can be anything (e.g. can_do).
372 See https://github.com/apache/airflow/issues/39144
373 :param resource_name: the name of the resource
374 :param user: the user to performing the action
375 """
377 @abstractmethod
378 def filter_authorized_menu_items(self, menu_items: list[MenuItem], *, user: T) -> list[MenuItem]:
379 """
380 Filter menu items based on user permissions.
382 :param menu_items: list of all menu items
383 :param user: the user
384 """
386 def is_authorized_hitl_task(self, *, assigned_users: set[str], user: T) -> bool:
387 """
388 Check if a user is allowed to approve/reject a HITL task.
390 Airflow only calls this method for tasks that have assigned users. When a task has none, Airflow
391 skips this method and any user allowed to update the task's HITL detail (``is_authorized_dag``
392 with ``DagAccessEntity.HITL_DETAIL``) can respond.
394 By default, checks if the user's ID is in the assigned_users set.
395 Auth managers can override this method to implement custom logic.
397 :param assigned_users: set of user IDs assigned to the task, never empty
398 :param user: the user to check authorization for
399 """
400 return user.get_id() in assigned_users
402 def batch_is_authorized_connection(
403 self,
404 requests: Sequence[IsAuthorizedConnectionRequest],
405 *,
406 user: T,
407 ) -> bool:
408 """
409 Batch version of ``is_authorized_connection``.
411 By default, calls individually the ``is_authorized_connection`` API on each item in the list of requests.
412 Can lead to some poor performance. It is recommended to override this method in the auth manager
413 implementation to provide a more efficient implementation.
415 :param requests: a list of requests containing the parameters for ``is_authorized_connection``
416 :param user: the user to performing the action
417 """
418 return all(
419 self.is_authorized_connection(
420 method=request["method"],
421 details=request.get("details"),
422 user=user,
423 )
424 for request in requests
425 )
427 def batch_is_authorized_dag(
428 self,
429 requests: Sequence[IsAuthorizedDagRequest],
430 *,
431 user: T,
432 ) -> bool:
433 """
434 Batch version of ``is_authorized_dag``.
436 By default, calls individually the ``is_authorized_dag`` API on each item in the list of requests.
437 Can lead to some poor performance. It is recommended to override this method in the auth manager
438 implementation to provide a more efficient implementation.
440 :param requests: a list of requests containing the parameters for ``is_authorized_dag``
441 :param user: the user to performing the action
442 """
443 return all(
444 self.is_authorized_dag(
445 method=request["method"],
446 access_entity=request.get("access_entity"),
447 details=request.get("details"),
448 user=user,
449 )
450 for request in requests
451 )
453 def batch_is_authorized_pool(
454 self,
455 requests: Sequence[IsAuthorizedPoolRequest],
456 *,
457 user: T,
458 ) -> bool:
459 """
460 Batch version of ``is_authorized_pool``.
462 By default, calls individually the ``is_authorized_pool`` API on each item in the list of requests.
463 Can lead to some poor performance. It is recommended to override this method in the auth manager
464 implementation to provide a more efficient implementation.
466 :param requests: a list of requests containing the parameters for ``is_authorized_pool``
467 :param user: the user to performing the action
468 """
469 return all(
470 self.is_authorized_pool(
471 method=request["method"],
472 details=request.get("details"),
473 user=user,
474 )
475 for request in requests
476 )
478 def batch_is_authorized_variable(
479 self,
480 requests: Sequence[IsAuthorizedVariableRequest],
481 *,
482 user: T,
483 ) -> bool:
484 """
485 Batch version of ``is_authorized_variable``.
487 By default, calls individually the ``is_authorized_variable`` API on each item in the list of requests.
488 Can lead to some poor performance. It is recommended to override this method in the auth manager
489 implementation to provide a more efficient implementation.
491 :param requests: a list of requests containing the parameters for ``is_authorized_variable``
492 :param user: the user to performing the action
493 """
494 return all(
495 self.is_authorized_variable(
496 method=request["method"],
497 details=request.get("details"),
498 user=user,
499 )
500 for request in requests
501 )
503 @provide_session
504 def get_authorized_connections(
505 self,
506 *,
507 user: T,
508 method: ResourceMethod = "GET",
509 session: Session = NEW_SESSION,
510 ) -> set[str]:
511 """
512 Get connection ids (``conn_id``) the user has access to.
514 :param user: the user
515 :param method: the method to filter on
516 :param session: the session
517 """
518 stmt = select(Connection.conn_id, Connection.team_name)
519 rows = session.execute(stmt).all()
520 connections_by_team: dict[str | None, set[str]] = defaultdict(set)
521 for conn_id, team_name in rows:
522 connections_by_team[team_name].add(conn_id)
524 conn_ids: set[str] = set()
525 for team_name, team_conn_ids in connections_by_team.items():
526 conn_ids.update(
527 self.filter_authorized_connections(
528 conn_ids=team_conn_ids, user=user, method=method, team_name=team_name
529 )
530 )
532 return conn_ids
534 def filter_authorized_connections(
535 self,
536 *,
537 conn_ids: set[str],
538 user: T,
539 method: ResourceMethod = "GET",
540 team_name: str | None = None,
541 ) -> set[str]:
542 """
543 Filter connections the user has access to.
545 By default, check individually if the user has permissions to access the connection.
546 Can lead to some poor performance. It is recommended to override this method in the auth manager
547 implementation to provide a more efficient implementation.
549 :param conn_ids: the set of connection ids (``conn_id``)
550 :param user: the user
551 :param method: the method to filter on
552 :param team_name: the name of the team associated to the connections if Airflow environment runs in
553 multi-team mode
554 """
556 def _is_authorized_connection(conn_id: str):
557 return self.is_authorized_connection(
558 method=method, details=ConnectionDetails(conn_id=conn_id, team_name=team_name), user=user
559 )
561 return {conn_id for conn_id in conn_ids if _is_authorized_connection(conn_id)}
563 @provide_session
564 def get_authorized_dag_ids(
565 self,
566 *,
567 user: T,
568 method: ResourceMethod = "GET",
569 session: Session = NEW_SESSION,
570 ) -> set[str]:
571 """
572 Get Dags the user has access to.
574 :param user: the user
575 :param method: the method to filter on
576 :param session: the session
577 """
578 stmt = (
579 select(DagModel.dag_id, dag_bundle_team_association_table.c.team_name)
580 .join(DagBundleModel, DagModel.bundle_name == DagBundleModel.name)
581 .join(
582 dag_bundle_team_association_table,
583 DagBundleModel.name == dag_bundle_team_association_table.c.dag_bundle_name,
584 isouter=True,
585 )
586 )
587 # The below type annotation is acceptable on SQLA2.1, but not on 2.0
588 rows: Sequence[Row[Unpack[tuple[str, str]]]] = session.execute(stmt).all() # type: ignore[type-arg]
589 dags_by_team: dict[str, set[str]] = defaultdict(set)
590 for dag_id, team_name in rows:
591 dags_by_team[team_name].add(dag_id)
593 dag_ids: set[str] = set()
594 for team_name, team_dag_ids in dags_by_team.items():
595 dag_ids.update(
596 self.filter_authorized_dag_ids(
597 dag_ids=team_dag_ids, user=user, method=method, team_name=team_name
598 )
599 )
601 return dag_ids
603 def filter_authorized_dag_ids(
604 self,
605 *,
606 dag_ids: set[str],
607 user: T,
608 method: ResourceMethod = "GET",
609 team_name: str | None = None,
610 ) -> set[str]:
611 """
612 Filter Dags the user has access to.
614 By default, check individually if the user has permissions to access the Dag.
615 Can lead to some poor performance. It is recommended to override this method in the auth manager
616 implementation to provide a more efficient implementation.
618 :param dag_ids: the set of Dag ids
619 :param user: the user
620 :param method: the method to filter on
621 :param team_name: the name of the team associated to the Dags if Airflow environment runs in
622 multi-team mode
623 """
625 def _is_authorized_dag_id(dag_id: str):
626 return self.is_authorized_dag(
627 method=method, details=DagDetails(id=dag_id, team_name=team_name), user=user
628 )
630 return {dag_id for dag_id in dag_ids if _is_authorized_dag_id(dag_id)}
632 @provide_session
633 def get_authorized_pools(
634 self,
635 *,
636 user: T,
637 method: ResourceMethod = "GET",
638 session: Session = NEW_SESSION,
639 ) -> set[str]:
640 """
641 Get pools the user has access to.
643 :param user: the user
644 :param method: the method to filter on
645 :param session: the session
646 """
647 stmt = select(Pool.pool, Pool.team_name)
648 rows = session.execute(stmt).all()
649 pools_by_team: dict[str | None, set[str]] = defaultdict(set)
650 for pool_name, team_name in rows:
651 pools_by_team[team_name].add(pool_name)
653 pool_names: set[str] = set()
654 for team_name, team_pool_names in pools_by_team.items():
655 pool_names.update(
656 self.filter_authorized_pools(
657 pool_names=team_pool_names, user=user, method=method, team_name=team_name
658 )
659 )
661 return pool_names
663 def filter_authorized_pools(
664 self,
665 *,
666 pool_names: set[str],
667 user: T,
668 method: ResourceMethod = "GET",
669 team_name: str | None = None,
670 ) -> set[str]:
671 """
672 Filter pools the user has access to.
674 By default, check individually if the user has permissions to access the pool.
675 Can lead to some poor performance. It is recommended to override this method in the auth manager
676 implementation to provide a more efficient implementation.
678 :param pool_names: the set of pool names
679 :param user: the user
680 :param method: the method to filter on
681 :param team_name: the name of the team associated to the connections if Airflow environment runs in
682 multi-team mode
683 """
685 def _is_authorized_pool(name: str):
686 return self.is_authorized_pool(
687 method=method, details=PoolDetails(name=name, team_name=team_name), user=user
688 )
690 return {pool_name for pool_name in pool_names if _is_authorized_pool(pool_name)}
692 @provide_session
693 def get_authorized_teams(
694 self,
695 *,
696 user: T,
697 method: ResourceMethod = "GET",
698 session: Session = NEW_SESSION,
699 ) -> set[str]:
700 """
701 Get teams the user belongs to.
703 :param user: the user
704 :param method: the method to filter on
705 :param session: the session
706 """
707 team_names = Team.get_all_team_names(session=session)
708 return self.filter_authorized_teams(teams_names=team_names, user=user, method=method)
710 def filter_authorized_teams(
711 self,
712 *,
713 teams_names: set[str],
714 user: T,
715 method: ResourceMethod = "GET",
716 ) -> set[str]:
717 """
718 Filter teams the user belongs to.
720 By default, check individually if the user has permissions to access the team.
721 Can lead to some poor performance. It is recommended to override this method in the auth manager
722 implementation to provide a more efficient implementation.
724 :param teams_names: the set of team names
725 :param user: the user
726 :param method: the method to filter on
727 """
729 def _is_authorized_team(name: str):
730 return self.is_authorized_team(method=method, details=TeamDetails(name=name), user=user)
732 return {team_name for team_name in teams_names if _is_authorized_team(team_name)}
734 @provide_session
735 def get_authorized_variables(
736 self,
737 *,
738 user: T,
739 method: ResourceMethod = "GET",
740 session: Session = NEW_SESSION,
741 ) -> set[str]:
742 """
743 Get variable keys the user has access to.
745 :param user: the user
746 :param method: the method to filter on
747 :param session: the session
748 """
749 stmt = select(Variable.key, Variable.team_name)
750 rows = session.execute(stmt).all()
751 variables_by_team: dict[str | None, set[str]] = defaultdict(set)
752 for var_key, team_name in rows:
753 variables_by_team[team_name].add(var_key)
755 var_keys: set[str] = set()
756 for team_name, team_var_keys in variables_by_team.items():
757 var_keys.update(
758 self.filter_authorized_variables(
759 variable_keys=team_var_keys, user=user, method=method, team_name=team_name
760 )
761 )
763 return var_keys
765 def filter_authorized_variables(
766 self,
767 *,
768 variable_keys: set[str],
769 user: T,
770 method: ResourceMethod = "GET",
771 team_name: str | None = None,
772 ) -> set[str]:
773 """
774 Filter variables the user has access to.
776 By default, check individually if the user has permissions to access the variable.
777 Can lead to some poor performance. It is recommended to override this method in the auth manager
778 implementation to provide a more efficient implementation.
780 :param variable_keys: the set of variable keys
781 :param user: the user
782 :param method: the method to filter on
783 :param team_name: the name of the team associated to the connections if Airflow environment runs in
784 multi-team mode
785 """
787 def _is_authorized_variable(var_key: str):
788 return self.is_authorized_variable(
789 method=method, details=VariableDetails(key=var_key, team_name=team_name), user=user
790 )
792 return {var_key for var_key in variable_keys if _is_authorized_variable(var_key)}
794 @staticmethod
795 def get_cli_commands() -> list[CLICommand]:
796 """
797 Vends CLI commands to be included in Airflow CLI.
799 Override this method to expose commands via Airflow CLI to manage this auth manager.
800 """
801 return []
803 def get_fastapi_app(self) -> FastAPI | None:
804 """
805 Specify a sub FastAPI application specific to the auth manager.
807 This sub application, if specified, is mounted in the main FastAPI application.
808 """
809 return None
811 def get_authorized_menu_items(self, *, user: T) -> list[MenuItem]:
812 """Get all menu items the user has access to."""
813 return self.filter_authorized_menu_items(list(MenuItem), user=user)
815 def get_extra_menu_items(self, *, user: T) -> list[ExtraMenuItem]:
816 """
817 Provide additional links to be added to the menu.
819 :param user: the user
820 """
821 return []
823 def _get_teams(self) -> set[str]:
824 """
825 Return the set of teams defined in the auth manager.
827 This method is used only when the Airflow environment is configured in multi-team mode.
828 """
829 raise NotImplementedError()
831 @staticmethod
832 def get_db_manager() -> str | None:
833 """
834 Specify the DB manager path needed to run the auth manager.
836 This is optional and not all auth managers require a DB manager.
837 """
838 return None
840 @staticmethod
841 def _get_jwt_audience() -> str:
842 """
843 Resolve the JWT audience from the documented ``[api_auth] jwt_audience`` option.
845 Falls back to the undocumented ``[api] jwt_audience`` location used by the signer in
846 earlier 3.x releases (with a deprecation warning) so deployments that set the wrong
847 section continue to work until they migrate. Returns the default ``apache-airflow``
848 when neither is configured.
850 :meta private:
851 """
852 if conf.has_option("api_auth", "jwt_audience"): 852 ↛ 853line 852 didn't jump to line 853 because the condition on line 852 was never true
853 return conf.get("api_auth", "jwt_audience")
854 if conf.has_option("api", "jwt_audience"): 854 ↛ 859line 854 didn't jump to line 859 because the condition on line 854 was never true
855 # Bug context in PR https://github.com/apache/airflow/pull/67494: the signer used to
856 # read `[api] jwt_audience` while the validator read `[api_auth] jwt_audience`, so
857 # any deployment that hit the bug set the value under `[api]`. Honour it with a
858 # deprecation warning until the fallback can be removed in a future major release.
859 warnings.warn(
860 "The `[api] jwt_audience` configuration option is deprecated and was never "
861 "documented. It was read only by the JWT signer due to a bug; the validator "
862 "always read `[api_auth] jwt_audience`. Move the value to `[api_auth] "
863 "jwt_audience` (env var `AIRFLOW__API_AUTH__JWT_AUDIENCE`). Support for the "
864 "`[api]` location will be removed in a future release.",
865 DeprecationWarning,
866 stacklevel=2,
867 )
868 return conf.get("api", "jwt_audience")
869 return "apache-airflow"
871 @classmethod
872 @cache
873 def _get_token_signer(
874 cls,
875 expiration_time_in_seconds: int = conf.getint("api_auth", "jwt_expiration_time"),
876 ) -> JWTGenerator:
877 """
878 Return the signer used to sign JWT token.
880 :meta private:
882 :param expiration_time_in_seconds: expiration time in seconds of the token
883 """
884 return JWTGenerator(
885 **get_signing_args(),
886 valid_for=expiration_time_in_seconds,
887 audience=cls._get_jwt_audience(),
888 )
890 @classmethod
891 @cache
892 def _get_token_validator(cls) -> JWTValidator:
893 """
894 Return the signer used to sign JWT token.
896 :meta private:
897 """
898 return JWTValidator(
899 **get_sig_validation_args(),
900 leeway=conf.getint("api_auth", "jwt_leeway"),
901 audience=cls._get_jwt_audience(),
902 )