/usr/lib/python2.7/dist-packages/google/auth/compute_engine/_metadata.py is in python-google-auth 1.3.0-2.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 | # Copyright 2016 Google Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Provides helper methods for talking to the Compute Engine metadata server.
See https://cloud.google.com/compute/docs/metadata for more details.
"""
import datetime
import json
import logging
import os
import six
from six.moves import http_client
from six.moves.urllib import parse as urlparse
from google.auth import _helpers
from google.auth import environment_vars
from google.auth import exceptions
_LOGGER = logging.getLogger(__name__)
_METADATA_ROOT = 'http://{}/computeMetadata/v1/'.format(
os.getenv(environment_vars.GCE_METADATA_ROOT, 'metadata.google.internal'))
# This is used to ping the metadata server, it avoids the cost of a DNS
# lookup.
_METADATA_IP_ROOT = 'http://{}'.format(
os.getenv(environment_vars.GCE_METADATA_IP, '169.254.169.254'))
_METADATA_FLAVOR_HEADER = 'metadata-flavor'
_METADATA_FLAVOR_VALUE = 'Google'
_METADATA_HEADERS = {_METADATA_FLAVOR_HEADER: _METADATA_FLAVOR_VALUE}
# Timeout in seconds to wait for the GCE metadata server when detecting the
# GCE environment.
try:
_METADATA_DEFAULT_TIMEOUT = int(os.getenv('GCE_METADATA_TIMEOUT', 3))
except ValueError: # pragma: NO COVER
_METADATA_DEFAULT_TIMEOUT = 3
def ping(request, timeout=_METADATA_DEFAULT_TIMEOUT):
"""Checks to see if the metadata server is available.
Args:
request (google.auth.transport.Request): A callable used to make
HTTP requests.
timeout (int): How long to wait for the metadata server to respond.
Returns:
bool: True if the metadata server is reachable, False otherwise.
"""
# NOTE: The explicit ``timeout`` is a workaround. The underlying
# issue is that resolving an unknown host on some networks will take
# 20-30 seconds; making this timeout short fixes the issue, but
# could lead to false negatives in the event that we are on GCE, but
# the metadata resolution was particularly slow. The latter case is
# "unlikely".
try:
response = request(
url=_METADATA_IP_ROOT, method='GET', headers=_METADATA_HEADERS,
timeout=timeout)
metadata_flavor = response.headers.get(_METADATA_FLAVOR_HEADER)
return (response.status == http_client.OK and
metadata_flavor == _METADATA_FLAVOR_VALUE)
except exceptions.TransportError:
_LOGGER.info('Compute Engine Metadata server unavailable.')
return False
def get(request, path, root=_METADATA_ROOT, recursive=False):
"""Fetch a resource from the metadata server.
Args:
request (google.auth.transport.Request): A callable used to make
HTTP requests.
path (str): The resource to retrieve. For example,
``'instance/service-accounts/default'``.
root (str): The full path to the metadata server root.
recursive (bool): Whether to do a recursive query of metadata. See
https://cloud.google.com/compute/docs/metadata#aggcontents for more
details.
Returns:
Union[Mapping, str]: If the metadata server returns JSON, a mapping of
the decoded JSON is return. Otherwise, the response content is
returned as a string.
Raises:
google.auth.exceptions.TransportError: if an error occurred while
retrieving metadata.
"""
base_url = urlparse.urljoin(root, path)
query_params = {}
if recursive:
query_params['recursive'] = 'true'
url = _helpers.update_query(base_url, query_params)
response = request(url=url, method='GET', headers=_METADATA_HEADERS)
if response.status == http_client.OK:
content = _helpers.from_bytes(response.data)
if response.headers['content-type'] == 'application/json':
try:
return json.loads(content)
except ValueError as caught_exc:
new_exc = exceptions.TransportError(
'Received invalid JSON from the Google Compute Engine'
'metadata service: {:.20}'.format(content))
six.raise_from(new_exc, caught_exc)
else:
return content
else:
raise exceptions.TransportError(
'Failed to retrieve {} from the Google Compute Engine'
'metadata service. Status: {} Response:\n{}'.format(
url, response.status, response.data), response)
def get_project_id(request):
"""Get the Google Cloud Project ID from the metadata server.
Args:
request (google.auth.transport.Request): A callable used to make
HTTP requests.
Returns:
str: The project ID
Raises:
google.auth.exceptions.TransportError: if an error occurred while
retrieving metadata.
"""
return get(request, 'project/project-id')
def get_service_account_info(request, service_account='default'):
"""Get information about a service account from the metadata server.
Args:
request (google.auth.transport.Request): A callable used to make
HTTP requests.
service_account (str): The string 'default' or a service account email
address. The determines which service account for which to acquire
information.
Returns:
Mapping: The service account's information, for example::
{
'email': '...',
'scopes': ['scope', ...],
'aliases': ['default', '...']
}
Raises:
google.auth.exceptions.TransportError: if an error occurred while
retrieving metadata.
"""
return get(
request,
'instance/service-accounts/{0}/'.format(service_account),
recursive=True)
def get_service_account_token(request, service_account='default'):
"""Get the OAuth 2.0 access token for a service account.
Args:
request (google.auth.transport.Request): A callable used to make
HTTP requests.
service_account (str): The string 'default' or a service account email
address. The determines which service account for which to acquire
an access token.
Returns:
Union[str, datetime]: The access token and its expiration.
Raises:
google.auth.exceptions.TransportError: if an error occurred while
retrieving metadata.
"""
token_json = get(
request,
'instance/service-accounts/{0}/token'.format(service_account))
token_expiry = _helpers.utcnow() + datetime.timedelta(
seconds=token_json['expires_in'])
return token_json['access_token'], token_expiry
|