fix: Rename aws imdsv2 url field and update token lifetime (#982)

* fix: Rename aws imdsv2 url field and update token lifetime

* update readme

* 🦉 Updates from OwlBot post-processor

See https://github.com/googleapis/repo-automation-bots/blob/main/packages/owl-bot/README.md

Co-authored-by: Owl Bot <gcf-owl-bot[bot]@users.noreply.github.com>

docs/user-guide.rst

@@ -261,7 +261,7 @@ following requirements are needed:
  external subject tokens and exchange them for service account access tokens.
 - If you want to use IDMSv2, then below field needs to be added to credential_source
  section of credential configuration.
- "aws_session_token_url": "http:https://169.254.169.254/latest/api/token"
+ "imdsv2_session_token_url": "http:https://169.254.169.254/latest/api/token"
 
 Follow the detailed instructions on how to
 `Configure Workload Identity Federation from AWS`_.

google/auth/aws.py

@@ -406,7 +406,9 @@ def __init__(
  self._cred_verification_url = credential_source.get(
  "regional_cred_verification_url"
  )
- self._aws_session_token_url = credential_source.get("aws_session_token_url")
+ self._imdsv2_session_token_url = credential_source.get(
+ "imdsv2_session_token_url"
+ )
  self._region = None
  self._request_signer = None
  self._target_resource = audience
@@ -460,32 +462,35 @@ def retrieve_subject_token(self, request):
  str: The retrieved subject token.
  """
  # Fetch the session token required to make meta data endpoint calls to aws
- if request is not None and self._aws_session_token_url is not None:
- headers = {"X-aws-ec2-metadata-token-ttl-seconds": "21600"}
+ if request is not None and self._imdsv2_session_token_url is not None:
+ headers = {"X-aws-ec2-metadata-token-ttl-seconds": "300"}
 
- session_token_response = request(
- url=self._aws_session_token_url, method="PUT", headers=headers
+ imdsv2_session_token_response = request(
+ url=self._imdsv2_session_token_url, method="PUT", headers=headers
  )
 
- if session_token_response.status != 200:
+ if imdsv2_session_token_response.status != 200:
  raise exceptions.RefreshError(
- "Unable to retrieve AWS Session Token", session_token_response.data
+ "Unable to retrieve AWS Session Token",
+ imdsv2_session_token_response.data,
  )
 
- session_token = session_token_response.data
+ imdsv2_session_token = imdsv2_session_token_response.data
  else:
- session_token = None
+ imdsv2_session_token = None
 
  # Initialize the request signer if not yet initialized after determining
  # the current AWS region.
  if self._request_signer is None:
- self._region = self._get_region(request, self._region_url, session_token)
+ self._region = self._get_region(
+ request, self._region_url, imdsv2_session_token
+ )
  self._request_signer = RequestSigner(self._region)
 
  # Retrieve the AWS security credentials needed to generate the signed
  # request.
  aws_security_credentials = self._get_security_credentials(
- request, session_token
+ request, imdsv2_session_token
  )
  # Generate the signed request to AWS STS GetCallerIdentity API.
  # Use the required regional endpoint. Otherwise, the request will fail.
@@ -531,15 +536,15 @@ def retrieve_subject_token(self, request):
  json.dumps(aws_signed_req, separators=(",", ":"), sort_keys=True)
  )
 
- def _get_region(self, request, url, session_token):
+ def _get_region(self, request, url, imdsv2_session_token):
  """Retrieves the current AWS region from either the AWS_REGION or
  AWS_DEFAULT_REGION environment variable or from the AWS metadata server.
 
  Args:
  request (google.auth.transport.Request): A callable used to make
  HTTP requests.
  url (str): The AWS metadata server region URL.
- session_token (str): The AWS session token to be added as a
+ imdsv2_session_token (str): The AWS IMDSv2 session token to be added as a
  header in the requests to AWS metadata endpoint.
 
  Returns:
@@ -564,8 +569,8 @@ def _get_region(self, request, url, session_token):
  raise exceptions.RefreshError("Unable to determine AWS region")
 
  headers = None
- if session_token is not None:
- headers = {"X-aws-ec2-metadata-token": session_token}
+ if imdsv2_session_token is not None:
+ headers = {"X-aws-ec2-metadata-token": imdsv2_session_token}
 
  response = request(url=self._region_url, method="GET", headers=headers)
 
@@ -585,15 +590,15 @@ def _get_region(self, request, url, session_token):
  # Only the us-east-2 part should be used.
  return response_body[:-1]
 
- def _get_security_credentials(self, request, session_token):
+ def _get_security_credentials(self, request, imdsv2_session_token):
  """Retrieves the AWS security credentials required for signing AWS
  requests from either the AWS security credentials environment variables
  or from the AWS metadata server.
 
  Args:
  request (google.auth.transport.Request): A callable used to make
  HTTP requests.
- session_token (str): The AWS session token to be added as a
+ imdsv2_session_token (str): The AWS IMDSv2 session token to be added as a
  header in the requests to AWS metadata endpoint.
 
  Returns:
@@ -620,11 +625,11 @@ def _get_security_credentials(self, request, session_token):
  }
 
  # Get role name.
- role_name = self._get_metadata_role_name(request, session_token)
+ role_name = self._get_metadata_role_name(request, imdsv2_session_token)
 
  # Get security credentials.
  credentials = self._get_metadata_security_credentials(
- request, role_name, session_token
+ request, role_name, imdsv2_session_token
  )
 
  return {
@@ -633,7 +638,9 @@ def _get_security_credentials(self, request, session_token):
  "security_token": credentials.get("Token"),
  }
 
- def _get_metadata_security_credentials(self, request, role_name, session_token):
+ def _get_metadata_security_credentials(
+ self, request, role_name, imdsv2_session_token
+ ):
  """Retrieves the AWS security credentials required for signing AWS
  requests from the AWS metadata server.
 
@@ -643,7 +650,7 @@ def _get_metadata_security_credentials(self, request, role_name, session_token):
  role_name (str): The AWS role name required by the AWS metadata
  server security_credentials endpoint in order to return the
  credentials.
- session_token (str): The AWS session token to be added as a
+ imdsv2_session_token (str): The AWS IMDSv2 session token to be added as a
  header in the requests to AWS metadata endpoint.
 
  Returns:
@@ -655,8 +662,8 @@ def _get_metadata_security_credentials(self, request, role_name, session_token):
  retrieving the AWS security credentials.
  """
  headers = {"Content-Type": "application/json"}
- if session_token is not None:
- headers["X-aws-ec2-metadata-token"] = session_token
+ if imdsv2_session_token is not None:
+ headers["X-aws-ec2-metadata-token"] = imdsv2_session_token
 
  response = request(
  url="{}/{}".format(self._security_credentials_url, role_name),
@@ -680,7 +687,7 @@ def _get_metadata_security_credentials(self, request, role_name, session_token):
 
  return credentials_response
 
- def _get_metadata_role_name(self, request, session_token):
+ def _get_metadata_role_name(self, request, imdsv2_session_token):
  """Retrieves the AWS role currently attached to the current AWS
  workload by querying the AWS metadata server. This is needed for the
  AWS metadata server security credentials endpoint in order to retrieve
@@ -689,7 +696,7 @@ def _get_metadata_role_name(self, request, session_token):
  Args:
  request (google.auth.transport.Request): A callable used to make
  HTTP requests.
- session_token (str): The AWS session token to be added as a
+ imdsv2_session_token (str): The AWS IMDSv2 session token to be added as a
  header in the requests to AWS metadata endpoint.
 
  Returns:
@@ -705,8 +712,8 @@ def _get_metadata_role_name(self, request, session_token):
  )
 
  headers = None
- if session_token is not None:
- headers = {"X-aws-ec2-metadata-token": session_token}
+ if imdsv2_session_token is not None:
+ headers = {"X-aws-ec2-metadata-token": imdsv2_session_token}
 
  response = request(
  url=self._security_credentials_url, method="GET", headers=headers

tests/test_aws.py

@@ -42,7 +42,7 @@
 SUBJECT_TOKEN_TYPE = "urn:ietf:params:aws:token-type:aws4_request"
 AUDIENCE = "//iam.googleapis.com/projects/123456/locations/global/workloadIdentityPools/POOL_ID/providers/PROVIDER_ID"
 REGION_URL = "http:https://169.254.169.254/latest/meta-data/placement/availability-zone"
-AWS_SESSION_TOKEN_URL = "http:https://169.254.169.254/latest/api/token"
+IMDSV2_SESSION_TOKEN_URL = "http:https://169.254.169.254/latest/api/token"
 SECURITY_CREDS_URL = "http:https://169.254.169.254/latest/meta-data/iam/security-credentials"
 CRED_VERIFICATION_URL = (
  "https://sts.{region}.amazonaws.com?Action=GetCallerIdentity&Version=2011-06-15"
@@ -579,7 +579,7 @@ class TestCredentials(object):
  "SecretAccessKey": SECRET_ACCESS_KEY,
  "Token": TOKEN,
  }
- AWS_SESSION_TOKEN = "awssessiontoken"
+ AWS_IMDSV2_SESSION_TOKEN = "awsimdsv2sessiontoken"
  AWS_SIGNATURE_TIME = "2020-08-11T06:55:22Z"
  CREDENTIAL_SOURCE = {
  "environment_id": "aws1",
@@ -656,21 +656,23 @@ def make_mock_request(
  token_data=None,
  impersonation_status=None,
  impersonation_data=None,
- session_token_status=None,
- session_token_data=None,
+ imdsv2_session_token_status=None,
+ imdsv2_session_token_data=None,
  ):
  """Utility function to generate a mock HTTP request object.
  This will facilitate testing various edge cases by specify how the
  various endpoints will respond while generating a Google Access token
  in an AWS environment.
  """
  responses = []
- if session_token_status:
+ if imdsv2_session_token_status:
  # AWS session token request
- session_response = mock.create_autospec(transport.Response, instance=True)
- session_response.status = session_token_status
- session_response.data = session_token_data
- responses.append(session_response)
+ imdsv2_session_response = mock.create_autospec(
+ transport.Response, instance=True
+ )
+ imdsv2_session_response.status = imdsv2_session_token_status
+ imdsv2_session_response.data = imdsv2_session_token_data
+ responses.append(imdsv2_session_response)
 
  if region_status:
  # AWS region request.
@@ -1035,11 +1037,13 @@ def test_retrieve_subject_token_success_temp_creds_no_environment_vars_idmsv2(
  role_name=self.AWS_ROLE,
  security_credentials_status=http_client.OK,
  security_credentials_data=self.AWS_SECURITY_CREDENTIALS_RESPONSE,
- session_token_status=http_client.OK,
- session_token_data=self.AWS_SESSION_TOKEN,
+ imdsv2_session_token_status=http_client.OK,
+ imdsv2_session_token_data=self.AWS_IMDSV2_SESSION_TOKEN,
  )
  credential_source_token_url = self.CREDENTIAL_SOURCE.copy()
- credential_source_token_url["aws_session_token_url"] = AWS_SESSION_TOKEN_URL
+ credential_source_token_url[
+ "imdsv2_session_token_url"
+ ] = IMDSV2_SESSION_TOKEN_URL
  credentials = self.make_credentials(
  credential_source=credential_source_token_url
  )
@@ -1056,29 +1060,29 @@ def test_retrieve_subject_token_success_temp_creds_no_environment_vars_idmsv2(
  # Assert session token request
  self.assert_aws_metadata_request_kwargs(
  request.call_args_list[0][1],
- AWS_SESSION_TOKEN_URL,
- {"X-aws-ec2-metadata-token-ttl-seconds": "21600"},
+ IMDSV2_SESSION_TOKEN_URL,
+ {"X-aws-ec2-metadata-token-ttl-seconds": "300"},
  "PUT",
  )
  # Assert region request.
  self.assert_aws_metadata_request_kwargs(
  request.call_args_list[1][1],
  REGION_URL,
- {"X-aws-ec2-metadata-token": self.AWS_SESSION_TOKEN},
+ {"X-aws-ec2-metadata-token": self.AWS_IMDSV2_SESSION_TOKEN},
  )
  # Assert role request.
  self.assert_aws_metadata_request_kwargs(
  request.call_args_list[2][1],
  SECURITY_CREDS_URL,
- {"X-aws-ec2-metadata-token": self.AWS_SESSION_TOKEN},
+ {"X-aws-ec2-metadata-token": self.AWS_IMDSV2_SESSION_TOKEN},
  )
  # Assert security credentials request.
  self.assert_aws_metadata_request_kwargs(
  request.call_args_list[3][1],
  "{}/{}".format(SECURITY_CREDS_URL, self.AWS_ROLE),
  {
  "Content-Type": "application/json",
- "X-aws-ec2-metadata-token": self.AWS_SESSION_TOKEN,
+ "X-aws-ec2-metadata-token": self.AWS_IMDSV2_SESSION_TOKEN,
  },
  )
 
@@ -1088,8 +1092,8 @@ def test_retrieve_subject_token_success_temp_creds_no_environment_vars_idmsv2(
  role_name=self.AWS_ROLE,
  security_credentials_status=http_client.OK,
  security_credentials_data=self.AWS_SECURITY_CREDENTIALS_RESPONSE,
- session_token_status=http_client.OK,
- session_token_data=self.AWS_SESSION_TOKEN,
+ imdsv2_session_token_status=http_client.OK,
+ imdsv2_session_token_data=self.AWS_IMDSV2_SESSION_TOKEN,
  )
 
  credentials.retrieve_subject_token(new_request)
@@ -1099,23 +1103,23 @@ def test_retrieve_subject_token_success_temp_creds_no_environment_vars_idmsv2(
  # Assert session token request
  self.assert_aws_metadata_request_kwargs(
  request.call_args_list[0][1],
- AWS_SESSION_TOKEN_URL,
- {"X-aws-ec2-metadata-token-ttl-seconds": "21600"},
+ IMDSV2_SESSION_TOKEN_URL,
+ {"X-aws-ec2-metadata-token-ttl-seconds": "300"},
  "PUT",
  )
  # Assert role request.
  self.assert_aws_metadata_request_kwargs(
  new_request.call_args_list[1][1],
  SECURITY_CREDS_URL,
- {"X-aws-ec2-metadata-token": self.AWS_SESSION_TOKEN},
+ {"X-aws-ec2-metadata-token": self.AWS_IMDSV2_SESSION_TOKEN},
  )
  # Assert security credentials request.
  self.assert_aws_metadata_request_kwargs(
  new_request.call_args_list[2][1],
  "{}/{}".format(SECURITY_CREDS_URL, self.AWS_ROLE),
  {
  "Content-Type": "application/json",
- "X-aws-ec2-metadata-token": self.AWS_SESSION_TOKEN,
+ "X-aws-ec2-metadata-token": self.AWS_IMDSV2_SESSION_TOKEN,
  },
  )
 
@@ -1125,11 +1129,13 @@ def test_retrieve_subject_token_session_error_idmsv2(self, utcnow):
  self.AWS_SIGNATURE_TIME, "%Y-%m-%dT%H:%M:%SZ"
  )
  request = self.make_mock_request(
- session_token_status=http_client.UNAUTHORIZED,
- session_token_data="unauthorized",
+ imdsv2_session_token_status=http_client.UNAUTHORIZED,
+ imdsv2_session_token_data="unauthorized",
  )
  credential_source_token_url = self.CREDENTIAL_SOURCE.copy()
- credential_source_token_url["aws_session_token_url"] = AWS_SESSION_TOKEN_URL
+ credential_source_token_url[
+ "imdsv2_session_token_url"
+ ] = IMDSV2_SESSION_TOKEN_URL
  credentials = self.make_credentials(
  credential_source=credential_source_token_url
  )
@@ -1142,8 +1148,8 @@ def test_retrieve_subject_token_session_error_idmsv2(self, utcnow):
  # Assert session token request
  self.assert_aws_metadata_request_kwargs(
  request.call_args_list[0][1],
- AWS_SESSION_TOKEN_URL,
- {"X-aws-ec2-metadata-token-ttl-seconds": "21600"},
+ IMDSV2_SESSION_TOKEN_URL,
+ {"X-aws-ec2-metadata-token-ttl-seconds": "300"},
  "PUT",
  )