Coverage for /home/airflow/.local/lib/python3.12/site-packages/airflow/api_fastapi/auth/managers/simple/services/login.py: 66%

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 hmac 

21 

22from fastapi import HTTPException, status 

23 

24from airflow.api_fastapi.app import get_auth_manager 

25from airflow.api_fastapi.auth.managers.simple.datamodels.login import LoginBody 

26from airflow.api_fastapi.auth.managers.simple.simple_auth_manager import SimpleAuthManager 

27from airflow.api_fastapi.auth.managers.simple.user import SimpleAuthManagerUser 

28from airflow.configuration import conf 

29 

30 

31class SimpleAuthManagerLogin: 

32 """Service for login.""" 

33 

34 @staticmethod 

35 def create_token( 

36 body: LoginBody, expiration_time_in_seconds: int = conf.getint("api_auth", "jwt_expiration_time") 

37 ) -> str: 

38 """ 

39 Authenticate user with given configuration. 

40 

41 :param body: LoginBody should include username and password 

42 :param expiration_time_in_seconds: int expiration time in seconds 

43 """ 

44 is_simple_auth_manager_all_admins = conf.getboolean("core", "simple_auth_manager_all_admins") 

45 if is_simple_auth_manager_all_admins: 45 ↛ 46line 45 didn't jump to line 46 because the condition on line 45 was never true

46 return SimpleAuthManagerLogin._create_anonymous_admin_user( 

47 expiration_time_in_seconds=expiration_time_in_seconds 

48 ) 

49 

50 if not body.username or not body.password: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true

51 raise HTTPException( 

52 status_code=status.HTTP_400_BAD_REQUEST, 

53 detail="Username and password must be provided", 

54 ) 

55 

56 users = SimpleAuthManager.get_users() 

57 passwords = SimpleAuthManager.get_passwords() 

58 # Use hmac.compare_digest for constant-time password comparison (CWE-208). 

59 # `passwords.get(..., "")` keeps the comparison constant-time against an 

60 # empty string when the user record exists but the password entry is missing. 

61 found_users = [ 

62 user 

63 for user in users 

64 if user.username == body.username 

65 and hmac.compare_digest( 

66 passwords.get(user.username, "").encode("utf-8"), 

67 body.password.encode("utf-8"), 

68 ) 

69 ] 

70 

71 if len(found_users) == 0: 71 ↛ 72line 71 didn't jump to line 72 because the condition on line 71 was never true

72 raise HTTPException( 

73 status_code=status.HTTP_401_UNAUTHORIZED, 

74 detail="Invalid credentials", 

75 ) 

76 

77 user = SimpleAuthManagerUser( 

78 username=body.username, 

79 role=found_users[0].role, 

80 teams=found_users[0].teams, 

81 ) 

82 

83 return get_auth_manager().generate_jwt( 

84 user=user, expiration_time_in_seconds=expiration_time_in_seconds 

85 ) 

86 

87 @staticmethod 

88 def create_token_all_admins( 

89 expiration_time_in_seconds: int = conf.getint("api_auth", "jwt_expiration_time"), 

90 ) -> str: 

91 is_simple_auth_manager_all_admins = conf.getboolean("core", "simple_auth_manager_all_admins") 

92 if not is_simple_auth_manager_all_admins: 

93 raise HTTPException( 

94 status.HTTP_403_FORBIDDEN, 

95 "This method is only allowed if ``[core] simple_auth_manager_all_admins`` is True", 

96 ) 

97 

98 return SimpleAuthManagerLogin._create_anonymous_admin_user( 

99 expiration_time_in_seconds=expiration_time_in_seconds 

100 ) 

101 

102 @staticmethod 

103 def _create_anonymous_admin_user( 

104 expiration_time_in_seconds: int = conf.getint("api_auth", "jwt_expiration_time"), 

105 ) -> str: 

106 user = SimpleAuthManagerUser( 

107 username="Anonymous", 

108 role="ADMIN", 

109 ) 

110 return get_auth_manager().generate_jwt( 

111 user=user, expiration_time_in_seconds=expiration_time_in_seconds 

112 )