Coverage for python / lsst / daf / butler / tests / server_utils.py: 0%

1# This file is part of daf_butler. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This software is dual licensed under the GNU General Public License and also 

10# under a 3-clause BSD license. Recipients may choose which of these licenses 

11# to use; please see the files gpl-3.0.txt and/or bsd_license.txt, 

12# respectively. If you choose the GPL option then the following text applies 

13# (but note that there is still no warranty even if you opt for BSD instead): 

14# 

15# This program is free software: you can redistribute it and/or modify 

16# it under the terms of the GNU General Public License as published by 

17# the Free Software Foundation, either version 3 of the License, or 

18# (at your option) any later version. 

19# 

20# This program is distributed in the hope that it will be useful, 

21# but WITHOUT ANY WARRANTY; without even the implied warranty of 

22# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

23# GNU General Public License for more details. 

24# 

25# You should have received a copy of the GNU General Public License 

26# along with this program. If not, see <http://www.gnu.org/licenses/>. 

27 

28 

29import re 

30from typing import Any 

31 

32from fastapi import FastAPI, HTTPException, Request, Response 

33 

34 

35def add_auth_header_check_middleware(app: FastAPI) -> None: 

36 """Add a middleware to a FastAPI app to check that Gafaelfawr 

37 authentication headers are present. 

38 

39 This is only suitable for testing -- in a real deployment, 

40 GafaelfawrIngress will handle these headers and convert them to a different 

41 format. 

42 

43 Parameters 

44 ---------- 

45 app : `FastAPI` 

46 The app the middleware will be added to. 

47 """ 

48 

49 @app.middleware("http") 

50 async def check_auth_headers(request: Request, call_next: Any) -> Response: 

51 if _is_authenticated_endpoint(request.url.path): 

52 header = request.headers.get("authorization") 

53 if header is None: 

54 raise HTTPException(status_code=401, detail="Authorization header is missing") 

55 if not re.match(r"^Bearer \S+", header): 

56 raise HTTPException( 

57 status_code=401, detail=f"Authorization header not in expected format: {header}" 

58 ) 

59 

60 return await call_next(request) 

61 

62 

63def _is_authenticated_endpoint(path: str) -> bool: 

64 """Return True if the specified path requires authentication in the real 

65 server deployment. 

66 """ 

67 if path == "/": 

68 return False 

69 if path.endswith("/butler.yaml"): 

70 return False 

71 if path.endswith("/butler.json"): 

72 return False 

73 if path.startswith("/api/butler/configs"): 

74 return False 

75 

76 return True