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

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 

20from fastapi import Depends, Request, status 

21from fastapi.responses import RedirectResponse 

22 

23from airflow.api_fastapi.app import get_cookie_path 

24from airflow.api_fastapi.auth.managers.base_auth_manager import COOKIE_NAME_JWT_TOKEN 

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

26from airflow.api_fastapi.auth.managers.simple.services.login import SimpleAuthManagerLogin 

27from airflow.api_fastapi.auth.managers.simple.utils import parse_login_body 

28from airflow.api_fastapi.common.router import AirflowRouter 

29from airflow.api_fastapi.common.types import Mimetype 

30from airflow.api_fastapi.core_api.openapi.exceptions import create_openapi_http_exception_doc 

31from airflow.api_fastapi.core_api.security import is_safe_url 

32from airflow.configuration import conf 

33 

34login_router = AirflowRouter(tags=["SimpleAuthManagerLogin"]) 

35 

36 

37@login_router.post( 

38 "/token", 

39 status_code=status.HTTP_201_CREATED, 

40 responses={ 

41 **create_openapi_http_exception_doc( 

42 [ 

43 status.HTTP_400_BAD_REQUEST, 

44 status.HTTP_401_UNAUTHORIZED, 

45 status.HTTP_415_UNSUPPORTED_MEDIA_TYPE, 

46 ] 

47 ), 

48 201: { 

49 "description": "Successful Response", 

50 "content": { 

51 Mimetype.JSON: {"schema": {"$ref": "#/components/schemas/LoginResponse"}}, 

52 }, 

53 }, 

54 }, 

55 openapi_extra={ 

56 "requestBody": { 

57 "required": True, 

58 "content": { 

59 "application/json": {"schema": {"$ref": "#/components/schemas/LoginBody"}}, 

60 "application/x-www-form-urlencoded": {"schema": {"$ref": "#/components/schemas/LoginBody"}}, 

61 }, 

62 } 

63 }, 

64) 

65def create_token( 

66 body: LoginBody = Depends(parse_login_body), 

67) -> LoginResponse: 

68 """Authenticate the user.""" 

69 return LoginResponse(access_token=SimpleAuthManagerLogin.create_token(body=body)) 

70 

71 

72@login_router.get( 

73 "/token", 

74 status_code=status.HTTP_201_CREATED, 

75 responses=create_openapi_http_exception_doc([status.HTTP_403_FORBIDDEN]), 

76) 

77def create_token_all_admins() -> LoginResponse: 

78 """Create a token with no credentials only if ``simple_auth_manager_all_admins`` is True.""" 

79 return LoginResponse(access_token=SimpleAuthManagerLogin.create_token_all_admins()) 

80 

81 

82@login_router.get( 

83 "/token/login", 

84 status_code=status.HTTP_307_TEMPORARY_REDIRECT, 

85 responses=create_openapi_http_exception_doc([status.HTTP_403_FORBIDDEN]), 

86) 

87def login_all_admins(request: Request) -> RedirectResponse: 

88 """Login the user with no credentials.""" 

89 fallback_url = conf.get("api", "base_url", fallback="/") 

90 next_url = request.query_params.get("next") 

91 redirect_url = next_url if next_url and is_safe_url(next_url, request=request) else fallback_url 

92 response = RedirectResponse(url=redirect_url) 

93 

94 # The default config has this as an empty string, so we can't use `has_option`. 

95 # And look at the request info (needs `--proxy-headers` flag to api-server) 

96 secure = request.base_url.scheme == "https" or bool(conf.get("api", "ssl_cert", fallback="")) 

97 

98 response.set_cookie( 

99 COOKIE_NAME_JWT_TOKEN, 

100 SimpleAuthManagerLogin.create_token_all_admins(), 

101 path=get_cookie_path(), 

102 secure=secure, 

103 httponly=True, 

104 samesite="lax", 

105 ) 

106 return response 

107 

108 

109@login_router.post( 

110 "/token/cli", 

111 status_code=status.HTTP_201_CREATED, 

112 responses=create_openapi_http_exception_doc([status.HTTP_400_BAD_REQUEST, status.HTTP_401_UNAUTHORIZED]), 

113) 

114def create_token_cli( 

115 body: LoginBody, 

116) -> LoginResponse: 

117 """Authenticate the user for the CLI.""" 

118 return LoginResponse( 

119 access_token=SimpleAuthManagerLogin.create_token( 

120 body=body, expiration_time_in_seconds=conf.getint("api_auth", "jwt_cli_expiration_time") 

121 ) 

122 )