Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/api_core/universe.py: 62%

1# Copyright 2024 Google LLC 

2# 

3# Licensed under the Apache License, Version 2.0 (the "License"); 

4# you may not use this file except in compliance with the License. 

5# You may obtain a copy of the License at 

6# 

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

8# 

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

10# distributed under the License is distributed on an "AS IS" BASIS, 

11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 

12# See the License for the specific language governing permissions and 

13# limitations under the License. 

14 

15"""Helpers for universe domain.""" 

16 

17from typing import Any, Optional 

18from urllib.parse import urlparse, urlunparse 

19 

20from google.auth.exceptions import MutualTLSChannelError # type: ignore 

21 

22DEFAULT_UNIVERSE = "googleapis.com" 

23 

24 

25class EmptyUniverseError(ValueError): 

26 def __init__(self): 

27 message = "Universe Domain cannot be an empty string." 

28 super().__init__(message) 

29 

30 

31class UniverseMismatchError(ValueError): 

32 def __init__(self, client_universe, credentials_universe): 

33 message = ( 

34 f"The configured universe domain ({client_universe}) does not match the universe domain " 

35 f"found in the credentials ({credentials_universe}). " 

36 "If you haven't configured the universe domain explicitly, " 

37 f"`{DEFAULT_UNIVERSE}` is the default." 

38 ) 

39 super().__init__(message) 

40 

41 

42def get_universe_domain( 

43 *potential_universes: Optional[str], 

44 default_universe: str, 

45) -> str: 

46 """Return the universe domain used by the client. 

47 

48 Args: 

49 *potential_universes (Optional[str]): Potential universe domains in order of preference. 

50 default_universe (str): The default universe domain. 

51 

52 Returns: 

53 str: The universe domain to be used by the client. 

54 

55 Raises: 

56 EmptyUniverseError: If the resolved universe domain is an empty string. 

57 """ 

58 resolved = next( 

59 (x.strip() for x in potential_universes if x is not None), 

60 default_universe, 

61 ) 

62 

63 if not resolved: 

64 raise EmptyUniverseError() 

65 return resolved 

66 

67 

68def determine_domain( 

69 client_universe_domain: Optional[str], universe_domain_env: Optional[str] 

70) -> str: 

71 """Return the universe domain used by the client. 

72 

73 Args: 

74 client_universe_domain (Optional[str]): The universe domain configured via the client options. 

75 universe_domain_env (Optional[str]): The universe domain configured via the 

76 "GOOGLE_CLOUD_UNIVERSE_DOMAIN" environment variable. 

77 

78 Returns: 

79 str: The universe domain to be used by the client. 

80 

81 Raises: 

82 ValueError: If the universe domain is an empty string. 

83 """ 

84 return get_universe_domain( 

85 client_universe_domain, 

86 universe_domain_env, 

87 default_universe=DEFAULT_UNIVERSE, 

88 ) 

89 

90 

91def compare_domains(client_universe: str, credentials: Any) -> bool: 

92 """Returns True iff the universe domains used by the client and credentials match. 

93 

94 Args: 

95 client_universe (str): The universe domain configured via the client options. 

96 credentials Any: The credentials being used in the client. 

97 

98 Returns: 

99 bool: True iff client_universe matches the universe in credentials. 

100 

101 Raises: 

102 ValueError: when client_universe does not match the universe in credentials. 

103 """ 

104 credentials_universe = getattr(credentials, "universe_domain", DEFAULT_UNIVERSE) 

105 

106 if client_universe != credentials_universe: 

107 raise UniverseMismatchError(client_universe, credentials_universe) 

108 return True 

109 

110 

111def get_default_mtls_endpoint(api_endpoint: Optional[str]) -> Optional[str]: 

112 """Converts api endpoint to mTLS endpoint. 

113 

114 Convert "*.sandbox.googleapis.com" and "*.googleapis.com" to 

115 "*.mtls.sandbox.googleapis.com" and "*.mtls.googleapis.com" respectively. 

116 Other URLs (including those that do not match these domain suffixes or 

117 already contain '.mtls.') are passed through as-is. 

118 

119 Args: 

120 api_endpoint (Optional[str]): the api endpoint to convert. 

121 

122 Returns: 

123 Optional[str]: converted mTLS api endpoint. 

124 """ 

125 if not api_endpoint or ".mtls." in api_endpoint.lower(): 

126 return api_endpoint 

127 

128 has_scheme = "://" in api_endpoint 

129 if not has_scheme: 

130 parsed = urlparse("//" + api_endpoint) 

131 else: 

132 parsed = urlparse(api_endpoint) 

133 

134 host = parsed.hostname 

135 if not host: 

136 return api_endpoint 

137 

138 port = f":{parsed.port}" if parsed.port else "" 

139 

140 lowered_host = host.lower() 

141 suffix_sandbox = ".sandbox.googleapis.com" 

142 suffix_google = ".googleapis.com" 

143 if lowered_host.endswith(suffix_sandbox): 

144 new_host = host[: -len(suffix_sandbox)] + ".mtls.sandbox.googleapis.com" 

145 elif lowered_host.endswith(suffix_google): 

146 new_host = host[: -len(suffix_google)] + ".mtls.googleapis.com" 

147 else: 

148 return api_endpoint 

149 

150 netloc = new_host + port 

151 new_parsed = parsed._replace(netloc=netloc) 

152 

153 if not has_scheme: 

154 return urlunparse(new_parsed)[2:] 

155 else: 

156 return urlunparse(new_parsed) 

157 

158 

159def get_api_endpoint( 

160 api_override: Optional[str], 

161 universe_domain: str, 

162 default_universe: str, 

163 default_mtls_endpoint: Optional[str], 

164 default_endpoint_template: str, 

165 use_mtls: bool, 

166) -> str: 

167 """Return the API endpoint used by the client. 

168 

169 Args: 

170 api_override (Optional[str]): The API endpoint override. If specified, 

171 this is always returned. 

172 universe_domain (str): The universe domain used by the client. 

173 default_universe (str): The default universe domain. 

174 default_mtls_endpoint (Optional[str]): The default mTLS endpoint. 

175 default_endpoint_template (str): The default endpoint template containing 

176 a placeholder `{UNIVERSE_DOMAIN}`. 

177 use_mtls (bool): Whether to use the mTLS endpoint. 

178 

179 Returns: 

180 str: The API endpoint to be used by the client. 

181 

182 Raises: 

183 google.auth.exceptions.MutualTLSChannelError: If mTLS is requested but 

184 not supported in the configured universe domain. 

185 ValueError: If mTLS is requested but no mTLS endpoint is available. 

186 """ 

187 if api_override is not None: 

188 return api_override 

189 

190 if use_mtls: 

191 if universe_domain.lower() != default_universe.lower(): 

192 raise MutualTLSChannelError( 

193 f"mTLS is not supported in any universe other than {default_universe}." 

194 ) 

195 if not default_mtls_endpoint: 

196 raise ValueError("mTLS endpoint is not available.") 

197 return default_mtls_endpoint 

198 else: 

199 return default_endpoint_template.format(UNIVERSE_DOMAIN=universe_domain)