Coverage for python / lsst / daf / butler / remote_butler / authentication / rubin.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 

28from __future__ import annotations 

29 

30import os 

31from fnmatch import fnmatchcase 

32from urllib.parse import urlparse 

33 

34from .interface import RemoteButlerAuthenticationProvider 

35 

36 

37class RubinAuthenticationProvider(RemoteButlerAuthenticationProvider): 

38 """Provide HTTP headers required for authenticating the user via Rubin 

39 Science Platform's Gafaelfawr service. 

40 

41 Parameters 

42 ---------- 

43 access_token : `str` 

44 Rubin Science Platform Gafaelfawr access token. 

45 """ 

46 

47 def __init__(self, access_token: str): 

48 # Access tokens are opaque bearer tokens. See https://sqr-069.lsst.io/ 

49 self._headers = {"Authorization": f"Bearer {access_token}"} 

50 

51 @staticmethod 

52 def create_from_environment(server_url: str) -> RubinAuthenticationProvider: 

53 access_token = _get_authentication_token_from_environment(server_url) 

54 if access_token is None: 

55 raise RuntimeError( 

56 "Attempting to connect to Butler server," 

57 " but no access credentials were found in the environment." 

58 ) 

59 return RubinAuthenticationProvider(access_token) 

60 

61 def get_server_headers(self) -> dict[str, str]: 

62 return self._headers 

63 

64 def get_datastore_headers(self) -> dict[str, str]: 

65 return {} 

66 

67 

68_SERVER_WHITELIST = ["*.lsst.cloud", "*.slac.stanford.edu"] 

69_EXPLICIT_BUTLER_ACCESS_TOKEN_ENVIRONMENT_KEY = "BUTLER_RUBIN_ACCESS_TOKEN" 

70_RSP_JUPYTER_ACCESS_TOKEN_ENVIRONMENT_KEY = "ACCESS_TOKEN" 

71 

72 

73def _get_authentication_token_from_environment(server_url: str) -> str | None: 

74 """Search the environment for a Rubin Science Platform access token. 

75 

76 The token may come from the following sources in this order: 

77 

78 1. The ``BUTLER_RUBIN_ACCESS_TOKEN`` environment variable. 

79 This environment variable is meant primarily for development use, 

80 running outside the Rubin Science Platform. This token will be sent 

81 to EVERY server that we connect to, so be careful when connecting to 

82 untrusted servers. 

83 2. The ``ACCESS_TOKEN`` environment variable. 

84 This environment variable is provided by the Rubin Science Platform 

85 Jupyter notebooks. It will only be returned if the given ``server_url`` 

86 is in a whitelist of servers known to belong to the Rubin Science 

87 Platform. Because this is a long-lived token that can be used to 

88 impersonate the user with their full access rights, it should not be 

89 sent to untrusted servers. 

90 

91 Parameters 

92 ---------- 

93 server_url : `str` 

94 URL of the Butler server that the caller intends to connect to. 

95 

96 Returns 

97 ------- 

98 access_token: `str` or `None` 

99 A Rubin Science Platform access token, or `None` if no token was 

100 configured in the environment. 

101 """ 

102 explicit_butler_token = os.getenv(_EXPLICIT_BUTLER_ACCESS_TOKEN_ENVIRONMENT_KEY) 

103 if explicit_butler_token: 

104 return explicit_butler_token 

105 

106 hostname = urlparse(server_url.lower()).hostname 

107 hostname_in_whitelist = any(hostname and fnmatchcase(hostname, pattern) for pattern in _SERVER_WHITELIST) 

108 notebook_token = os.getenv(_RSP_JUPYTER_ACCESS_TOKEN_ENVIRONMENT_KEY) 

109 if hostname_in_whitelist and notebook_token: 

110 return notebook_token 

111 

112 return None