Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/core_api/routes/ui/auth.py: 59%

1# Licensed to the Apache Software Foundation (ASF) under one 

2# or more contributor license agreements. See the NOTICE file 

3# distributed with this work for additional information 

4# regarding copyright ownership. The ASF licenses this file 

5# to you under the Apache License, Version 2.0 (the 

6# "License"); you may not use this file except in compliance 

7# with the License. You may obtain a copy of the License at 

8# 

9# http://www.apache.org/licenses/LICENSE-2.0 

10# 

11# Unless required by applicable law or agreed to in writing, 

12# software distributed under the License is distributed on an 

13# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY 

14# KIND, either express or implied. See the License for the 

15# specific language governing permissions and limitations 

16# under the License. 

17 

18from __future__ import annotations 

19 

20import logging 

21 

22from fastapi import Depends 

23 

24from airflow.api_fastapi.app import get_auth_manager 

25from airflow.api_fastapi.common.router import AirflowRouter 

26from airflow.api_fastapi.core_api.datamodels.ui.auth import ( 

27 AuthenticatedMeResponse, 

28 GenerateTokenBody, 

29 GenerateTokenResponse, 

30 MenuItemCollectionResponse, 

31 TokenType, 

32) 

33from airflow.api_fastapi.core_api.security import GetUserDep 

34from airflow.api_fastapi.logging.decorators import action_logging 

35from airflow.configuration import conf 

36 

37log = logging.getLogger(__name__) 

38 

39auth_router = AirflowRouter(tags=["Auth Links"]) 

40 

41 

42@auth_router.get("/auth/menus") 

43def get_auth_menus( 

44 user: GetUserDep, 

45) -> MenuItemCollectionResponse: 

46 authorized_menu_items = get_auth_manager().get_authorized_menu_items(user=user) 

47 extra_menu_items = get_auth_manager().get_extra_menu_items(user=user) 

48 

49 return MenuItemCollectionResponse( 

50 authorized_menu_items=authorized_menu_items, 

51 extra_menu_items=extra_menu_items, 

52 ) 

53 

54 

55@auth_router.get("/auth/me") 

56def get_current_user_info( 

57 user: GetUserDep, 

58) -> AuthenticatedMeResponse: 

59 """Convienently get the current authenticated user information.""" 

60 return AuthenticatedMeResponse( 

61 id=user.get_id(), 

62 username=user.get_name(), 

63 ) 

64 

65 

66@auth_router.post("/auth/token", dependencies=[Depends(action_logging())]) 

67def generate_token( 

68 body: GenerateTokenBody, 

69 user: GetUserDep, 

70) -> GenerateTokenResponse: 

71 """Generate a JWT token for the authenticated user.""" 

72 if body.token_type == TokenType.CLI: 

73 expiration_seconds = conf.getint("api_auth", "jwt_cli_expiration_time") 

74 else: 

75 expiration_seconds = conf.getint("api_auth", "jwt_expiration_time") 

76 

77 access_token = get_auth_manager().generate_jwt(user, expiration_time_in_seconds=expiration_seconds) 

78 

79 log.info( 

80 "User %s generated a %s token (expires in %d seconds)", 

81 user.get_name(), 

82 body.token_type.value, 

83 expiration_seconds, 

84 ) 

85 

86 return GenerateTokenResponse( 

87 access_token=access_token, 

88 token_type=body.token_type, 

89 expires_in_seconds=expiration_seconds, 

90 )