import os

from urllib.parse import urljoin

import requests

from google.auth import _helpers

from google.auth import environment_vars

from google.auth import exceptions

from google.auth import metrics

from google.auth import transport

from google.auth._exponential_backoff import ExponentialBackoff

from google.auth.compute_engine import _mtls

_LOGGER = logging.getLogger(__name__)

_GCE_DEFAULT_MDS_IP = "169.254.169.254"

_GCE_DEFAULT_HOST = "metadata.google.internal"

_GCE_DEFAULT_MDS_HOSTS = [_GCE_DEFAULT_HOST, _GCE_DEFAULT_MDS_IP]

# Environment variable GCE_METADATA_HOST is originally named

# GCE_METADATA_ROOT. For compatibility reasons, here it checks

# the new variable first; if not set, the system falls back

# to the old variable.

_GCE_METADATA_HOST = os.getenv(environment_vars.GCE_METADATA_HOST, None)

if not _GCE_METADATA_HOST:

_GCE_METADATA_HOST = os.getenv(

environment_vars.GCE_METADATA_ROOT, "metadata.google.internal"

environment_vars.GCE_METADATA_ROOT, _GCE_DEFAULT_HOST

)

def _validate_gce_mds_configured_environment():

"""Validates the GCE metadata server environment configuration for mTLS.

mTLS is only supported when connecting to the default metadata server hosts.

If we are in strict mode (which requires mTLS), ensure that the metadata host

has not been overridden to a custom value (which means mTLS will fail).

Raises:

google.auth.exceptions.MutualTLSChannelError: if the environment

configuration is invalid for mTLS.

"""

mode = _mtls._parse_mds_mode()

if mode == _mtls.MdsMtlsMode.STRICT:

# mTLS is only supported when connecting to the default metadata host.

# Raise an exception if we are in strict mode (which requires mTLS)

# but the metadata host has been overridden to a custom MDS. (which means mTLS will fail)

if _GCE_METADATA_HOST not in _GCE_DEFAULT_MDS_HOSTS:

raise exceptions.MutualTLSChannelError(

"Mutual TLS is required, but the metadata host has been overridden. "

"mTLS is only supported when connecting to the default metadata host."

)

def _get_metadata_root(use_mtls: bool):

"""Returns the metadata server root URL."""

scheme = "https" if use_mtls else "http"

return "{}://{}/computeMetadata/v1/".format(scheme, _GCE_METADATA_HOST)

def _get_metadata_ip_root(use_mtls: bool):

"""Returns the metadata server IP root URL."""

scheme = "https" if use_mtls else "http"

return "{}://{}".format(

scheme, os.getenv(environment_vars.GCE_METADATA_IP, _GCE_DEFAULT_MDS_IP)

)

_METADATA_ROOT = "http://{}/computeMetadata/v1/".format(_GCE_METADATA_HOST)

# 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}

return content.startswith(_GOOGLE)

def _prepare_request_for_mds(request, use_mtls=False) -> None:

"""Prepares a request for the metadata server.

This will check if mTLS should be used and mount the mTLS adapter if needed.

Args:

request (google.auth.transport.Request): A callable used to make

HTTP requests.

use_mtls (bool): Whether to use mTLS for the request.

Returns:

google.auth.transport.Request: A request object to use.

If mTLS is enabled, the request will have the mTLS adapter mounted.

Otherwise, the original request will be returned unchanged.

"""

# Only modify the request if mTLS is enabled.

if use_mtls:

# Ensure the request has a session to mount the adapter to.

if not request.session:

request.session = requests.Session()

adapter = _mtls.MdsMtlsAdapter()

# Mount the adapter for all default GCE metadata hosts.

for host in _GCE_DEFAULT_MDS_HOSTS:

request.session.mount(f"https://{host}/", adapter)

def ping(request, timeout=_METADATA_DEFAULT_TIMEOUT, retry_count=3):

"""Checks to see if the metadata server is available.

Returns:

bool: True if the metadata server is reachable, False otherwise.

"""

use_mtls = _mtls.should_use_mds_mtls()

_prepare_request_for_mds(request, use_mtls=use_mtls)

# 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

for attempt in backoff:

try:

response = request(

url=_METADATA_IP_ROOT, method="GET", headers=headers, timeout=timeout

url=_get_metadata_ip_root(use_mtls),

method="GET",

headers=headers,

timeout=timeout,

)

metadata_flavor = response.headers.get(_METADATA_FLAVOR_HEADER)

def get(

request,

path,

root=_METADATA_ROOT,

root=None,

params=None,

recursive=False,

retry_count=5,

HTTP requests.

path (str): The resource to retrieve. For example,

``'instance/service-accounts/default'``.

root (str): The full path to the metadata server root.

root (Optional[str]): The full path to the metadata server root. If not

provided, the default root will be used.

params (Optional[Mapping[str, str]]): A mapping of query parameter

keys to values.

recursive (bool): Whether to do a recursive query of metadata. See

Raises:

google.auth.exceptions.TransportError: if an error occurred while

retrieving metadata.

google.auth.exceptions.MutualTLSChannelError: if using mtls and the environment

configuration is invalid for mTLS (for example, the metadata host

has been overridden in strict mTLS mode).

"""

use_mtls = _mtls.should_use_mds_mtls()

# Prepare the request object for mTLS if needed.

# This will create a new request object with the mTLS session.

_prepare_request_for_mds(request, use_mtls=use_mtls)

if root is None:

root = _get_metadata_root(use_mtls)

# mTLS is only supported when connecting to the default metadata host.

# If we are in strict mode (which requires mTLS), ensure that the metadata host

# has not been overridden to a non-default host value (which means mTLS will fail).

_validate_gce_mds_configured_environment()

base_url = urljoin(root, path)

query_params = {} if params is None else params