Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/google/auth/_helpers.py: 31%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1# Copyright 2015 Google Inc.
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.
15"""Helper functions for commonly used utilities."""
17import base64
18import calendar
19import datetime
20import hashlib
21import json
22import logging
23import sys
24import urllib
25from email.message import Message
26from typing import Any, Dict, Mapping, Optional, Union
28from google.auth import exceptions
30DEFAULT_UNIVERSE_DOMAIN = "googleapis.com"
32# _BASE_LOGGER_NAME is the base logger for all google-based loggers.
33_BASE_LOGGER_NAME = "google"
35# _LOGGING_INITIALIZED ensures that base logger is only configured once
36# (unless already configured by the end-user).
37_LOGGING_INITIALIZED = False
40# The smallest MDS cache used by this library stores tokens until 4 minutes from
41# expiry.
42REFRESH_THRESHOLD = datetime.timedelta(minutes=3, seconds=45)
44# TODO(https://github.com/googleapis/google-auth-library-python/issues/1684): Audit and update the list below.
45_SENSITIVE_FIELDS = {
46 "accessToken",
47 "access_token",
48 "id_token",
49 "client_id",
50 "refresh_token",
51 "client_secret",
52}
55def copy_docstring(source_class):
56 """Decorator that copies a method's docstring from another class.
58 Args:
59 source_class (type): The class that has the documented method.
61 Returns:
62 Callable: A decorator that will copy the docstring of the same
63 named method in the source class to the decorated method.
64 """
66 def decorator(method):
67 """Decorator implementation.
69 Args:
70 method (Callable): The method to copy the docstring to.
72 Returns:
73 Callable: the same method passed in with an updated docstring.
75 Raises:
76 google.auth.exceptions.InvalidOperation: if the method already has a docstring.
77 """
78 if method.__doc__:
79 raise exceptions.InvalidOperation("Method already has a docstring.")
81 source_method = getattr(source_class, method.__name__)
82 method.__doc__ = source_method.__doc__
84 return method
86 return decorator
89def parse_content_type(header_value):
90 """Parse a 'content-type' header value to get just the plain media-type (without parameters).
92 This is done using the class Message from email.message as suggested in PEP 594
93 (because the cgi is now deprecated and will be removed in python 3.13,
94 see https://peps.python.org/pep-0594/#cgi).
96 Args:
97 header_value (str): The value of a 'content-type' header as a string.
99 Returns:
100 str: A string with just the lowercase media-type from the parsed 'content-type' header.
101 If the provided content-type is not parsable, returns 'text/plain',
102 the default value for textual files.
103 """
104 m = Message()
105 m["content-type"] = header_value
106 return (
107 m.get_content_type()
108 ) # Despite the name, actually returns just the media-type
111def utcnow():
112 """Returns the current UTC datetime.
114 Returns:
115 datetime: The current time in UTC.
116 """
117 # We used datetime.utcnow() before, since it's deprecated from python 3.12,
118 # we are using datetime.now(timezone.utc) now. "utcnow()" is offset-native
119 # (no timezone info), but "now()" is offset-aware (with timezone info).
120 # This will cause datetime comparison problem. For backward compatibility,
121 # we need to remove the timezone info.
122 now = datetime.datetime.now(datetime.timezone.utc)
123 now = now.replace(tzinfo=None)
124 return now
127def utcfromtimestamp(timestamp):
128 """Returns the UTC datetime from a timestamp.
130 Args:
131 timestamp (float): The timestamp to convert.
133 Returns:
134 datetime: The time in UTC.
135 """
136 # We used datetime.utcfromtimestamp() before, since it's deprecated from
137 # python 3.12, we are using datetime.fromtimestamp(timestamp, timezone.utc)
138 # now. "utcfromtimestamp()" is offset-native (no timezone info), but
139 # "fromtimestamp(timestamp, timezone.utc)" is offset-aware (with timezone
140 # info). This will cause datetime comparison problem. For backward
141 # compatibility, we need to remove the timezone info.
142 dt = datetime.datetime.fromtimestamp(timestamp, tz=datetime.timezone.utc)
143 dt = dt.replace(tzinfo=None)
144 return dt
147def datetime_to_secs(value):
148 """Convert a datetime object to the number of seconds since the UNIX epoch.
150 Args:
151 value (datetime): The datetime to convert.
153 Returns:
154 int: The number of seconds since the UNIX epoch.
155 """
156 return calendar.timegm(value.utctimetuple())
159def to_bytes(value, encoding="utf-8"):
160 """Converts a string value to bytes, if necessary.
162 Args:
163 value (Union[str, bytes]): The value to be converted.
164 encoding (str): The encoding to use to convert unicode to bytes.
165 Defaults to "utf-8".
167 Returns:
168 bytes: The original value converted to bytes (if unicode) or as
169 passed in if it started out as bytes.
171 Raises:
172 google.auth.exceptions.InvalidValue: If the value could not be converted to bytes.
173 """
174 result = value.encode(encoding) if isinstance(value, str) else value
175 if isinstance(result, bytes):
176 return result
177 else:
178 raise exceptions.InvalidValue(
179 "{0!r} could not be converted to bytes".format(value)
180 )
183def from_bytes(value):
184 """Converts bytes to a string value, if necessary.
186 Args:
187 value (Union[str, bytes]): The value to be converted.
189 Returns:
190 str: The original value converted to unicode (if bytes) or as passed in
191 if it started out as unicode.
193 Raises:
194 google.auth.exceptions.InvalidValue: If the value could not be converted to unicode.
195 """
196 result = value.decode("utf-8") if isinstance(value, bytes) else value
197 if isinstance(result, str):
198 return result
199 else:
200 raise exceptions.InvalidValue(
201 "{0!r} could not be converted to unicode".format(value)
202 )
205def update_query(url, params, remove=None):
206 """Updates a URL's query parameters.
208 Replaces any current values if they are already present in the URL.
210 Args:
211 url (str): The URL to update.
212 params (Mapping[str, str]): A mapping of query parameter
213 keys to values.
214 remove (Sequence[str]): Parameters to remove from the query string.
216 Returns:
217 str: The URL with updated query parameters.
219 Examples:
221 >>> url = 'http://example.com?a=1'
222 >>> update_query(url, {'a': '2'})
223 http://example.com?a=2
224 >>> update_query(url, {'b': '3'})
225 http://example.com?a=1&b=3
226 >> update_query(url, {'b': '3'}, remove=['a'])
227 http://example.com?b=3
229 """
230 if remove is None:
231 remove = []
233 # Split the URL into parts.
234 parts = urllib.parse.urlparse(url)
235 # Parse the query string.
236 query_params = urllib.parse.parse_qs(parts.query)
237 # Update the query parameters with the new parameters.
238 query_params.update(params)
239 # Remove any values specified in remove.
240 query_params = {
241 key: value for key, value in query_params.items() if key not in remove
242 }
243 # Re-encoded the query string.
244 new_query = urllib.parse.urlencode(query_params, doseq=True)
245 # Unsplit the url.
246 new_parts = parts._replace(query=new_query)
247 return urllib.parse.urlunparse(new_parts)
250def scopes_to_string(scopes):
251 """Converts scope value to a string suitable for sending to OAuth 2.0
252 authorization servers.
254 Args:
255 scopes (Sequence[str]): The sequence of scopes to convert.
257 Returns:
258 str: The scopes formatted as a single string.
259 """
260 return " ".join(scopes)
263def string_to_scopes(scopes):
264 """Converts stringifed scopes value to a list.
266 Args:
267 scopes (Union[Sequence, str]): The string of space-separated scopes
268 to convert.
269 Returns:
270 Sequence(str): The separated scopes.
271 """
272 if not scopes:
273 return []
275 return scopes.split(" ")
278def padded_urlsafe_b64decode(value):
279 """Decodes base64 strings lacking padding characters.
281 Google infrastructure tends to omit the base64 padding characters.
283 Args:
284 value (Union[str, bytes]): The encoded value.
286 Returns:
287 bytes: The decoded value
288 """
289 b64string = to_bytes(value)
290 padded = b64string + b"=" * (-len(b64string) % 4)
291 return base64.urlsafe_b64decode(padded)
294def unpadded_urlsafe_b64encode(value):
295 """Encodes base64 strings removing any padding characters.
297 `rfc 7515`_ defines Base64url to NOT include any padding
298 characters, but the stdlib doesn't do that by default.
300 _rfc7515: https://tools.ietf.org/html/rfc7515#page-6
302 Args:
303 value (Union[str|bytes]): The bytes-like value to encode
305 Returns:
306 Union[str|bytes]: The encoded value
307 """
308 return base64.urlsafe_b64encode(value).rstrip(b"=")
311def is_python_3():
312 """Check if the Python interpreter is Python 2 or 3.
314 Returns:
315 bool: True if the Python interpreter is Python 3 and False otherwise.
316 """
318 return sys.version_info > (3, 0) # pragma: NO COVER
321def _hash_sensitive_info(data: Union[dict, list]) -> Union[dict, list, str]:
322 """
323 Hashes sensitive information within a dictionary.
325 Args:
326 data: The dictionary containing data to be processed.
328 Returns:
329 A new dictionary with sensitive values replaced by their SHA512 hashes.
330 If the input is a list, returns a list with each element recursively processed.
331 If the input is neither a dict nor a list, returns the type of the input as a string.
333 """
334 if isinstance(data, dict):
335 hashed_data: Dict[Any, Union[Optional[str], dict, list]] = {}
336 for key, value in data.items():
337 if key in _SENSITIVE_FIELDS and not isinstance(value, (dict, list)):
338 hashed_data[key] = _hash_value(value, key)
339 elif isinstance(value, (dict, list)):
340 hashed_data[key] = _hash_sensitive_info(value)
341 else:
342 hashed_data[key] = value
343 return hashed_data
344 elif isinstance(data, list):
345 hashed_list = []
346 for val in data:
347 hashed_list.append(_hash_sensitive_info(val))
348 return hashed_list
349 else:
350 # TODO(https://github.com/googleapis/google-auth-library-python/issues/1701):
351 # Investigate and hash sensitive info before logging when the data type is
352 # not a dict or a list.
353 return str(type(data))
356def _hash_value(value, field_name: str) -> Optional[str]:
357 """Hashes a value and returns a formatted hash string."""
358 if value is None:
359 return None
360 encoded_value = str(value).encode("utf-8")
361 hash_object = hashlib.sha512()
362 hash_object.update(encoded_value)
363 hex_digest = hash_object.hexdigest()
364 return f"hashed_{field_name}-{hex_digest}"
367def _logger_configured(logger: logging.Logger) -> bool:
368 """Determines whether `logger` has non-default configuration
370 Args:
371 logger: The logger to check.
373 Returns:
374 bool: Whether the logger has any non-default configuration.
375 """
376 return (
377 logger.handlers != [] or logger.level != logging.NOTSET or not logger.propagate
378 )
381def is_logging_enabled(logger: logging.Logger) -> bool:
382 """
383 Checks if debug logging is enabled for the given logger.
385 Args:
386 logger: The logging.Logger instance to check.
388 Returns:
389 True if debug logging is enabled, False otherwise.
390 """
391 # NOTE: Log propagation to the root logger is disabled unless
392 # the base logger i.e. logging.getLogger("google") is
393 # explicitly configured by the end user. Ideally this
394 # needs to happen in the client layer (already does for GAPICs).
395 # However, this is implemented here to avoid logging
396 # (if a root logger is configured) when a version of google-auth
397 # which supports logging is used with:
398 # - an older version of a GAPIC which does not support logging.
399 # - Apiary client which does not support logging.
400 global _LOGGING_INITIALIZED
401 if not _LOGGING_INITIALIZED:
402 base_logger = logging.getLogger(_BASE_LOGGER_NAME)
403 if not _logger_configured(base_logger):
404 base_logger.propagate = False
405 _LOGGING_INITIALIZED = True
407 return logger.isEnabledFor(logging.DEBUG)
410def request_log(
411 logger: logging.Logger,
412 method: str,
413 url: str,
414 body: Optional[bytes],
415 headers: Optional[Mapping[str, str]],
416) -> None:
417 """
418 Logs an HTTP request at the DEBUG level if logging is enabled.
420 Args:
421 logger: The logging.Logger instance to use.
422 method: The HTTP method (e.g., "GET", "POST").
423 url: The URL of the request.
424 body: The request body (can be None).
425 headers: The request headers (can be None).
426 """
427 if is_logging_enabled(logger):
428 content_type = (
429 headers["Content-Type"] if headers and "Content-Type" in headers else ""
430 )
431 json_body = _parse_request_body(body, content_type=content_type)
432 logged_body = _hash_sensitive_info(json_body)
433 logger.debug(
434 "Making request...",
435 extra={
436 "httpRequest": {
437 "method": method,
438 "url": url,
439 "body": logged_body,
440 "headers": headers,
441 }
442 },
443 )
446def _parse_request_body(body: Optional[bytes], content_type: str = "") -> Any:
447 """
448 Parses a request body, handling bytes and string types, and different content types.
450 Args:
451 body (Optional[bytes]): The request body.
452 content_type (str): The content type of the request body, e.g., "application/json",
453 "application/x-www-form-urlencoded", or "text/plain". If empty, attempts
454 to parse as JSON.
456 Returns:
457 Parsed body (dict, str, or None).
458 - JSON: Decodes if content_type is "application/json" or None (fallback).
459 - URL-encoded: Parses if content_type is "application/x-www-form-urlencoded".
460 - Plain text: Returns string if content_type is "text/plain".
461 - None: Returns if body is None, UTF-8 decode fails, or content_type is unknown.
462 """
463 if body is None:
464 return None
465 try:
466 body_str = body.decode("utf-8")
467 except (UnicodeDecodeError, AttributeError):
468 return None
469 content_type = content_type.lower()
470 if not content_type or "application/json" in content_type:
471 try:
472 return json.loads(body_str)
473 except (TypeError, ValueError):
474 return body_str
475 if "application/x-www-form-urlencoded" in content_type:
476 parsed_query = urllib.parse.parse_qs(body_str)
477 result = {k: v[0] for k, v in parsed_query.items()}
478 return result
479 if "text/plain" in content_type:
480 return body_str
481 return None
484def _parse_response(response: Any) -> Any:
485 """
486 Parses a response, attempting to decode JSON.
488 Args:
489 response: The response object to parse. This can be any type, but
490 it is expected to have a `json()` method if it contains JSON.
492 Returns:
493 The parsed response. If the response contains valid JSON, the
494 decoded JSON object (e.g., a dictionary or list) is returned.
495 If the response does not have a `json()` method or if the JSON
496 decoding fails, None is returned.
497 """
498 try:
499 json_response = response.json()
500 return json_response
501 except Exception:
502 # TODO(https://github.com/googleapis/google-auth-library-python/issues/1744):
503 # Parse and return response payload as json based on different content types.
504 return None
507def _response_log_base(logger: logging.Logger, parsed_response: Any) -> None:
508 """
509 Logs a parsed HTTP response at the DEBUG level.
511 This internal helper function takes a parsed response and logs it
512 using the provided logger. It also applies a hashing function to
513 potentially sensitive information before logging.
515 Args:
516 logger: The logging.Logger instance to use for logging.
517 parsed_response: The parsed HTTP response object (e.g., a dictionary,
518 list, or the original response if parsing failed).
519 """
521 logged_response = _hash_sensitive_info(parsed_response)
522 logger.debug("Response received...", extra={"httpResponse": logged_response})
525def response_log(logger: logging.Logger, response: Any) -> None:
526 """
527 Logs an HTTP response at the DEBUG level if logging is enabled.
529 Args:
530 logger: The logging.Logger instance to use.
531 response: The HTTP response object to log.
532 """
533 if is_logging_enabled(logger):
534 json_response = _parse_response(response)
535 _response_log_base(logger, json_response)