NO_URLS_MATCHED_PREFIX = 'No URLs matched'
NO_URLS_MATCHED_GENERIC = (NO_URLS_MATCHED_PREFIX +
                           '. Do the files you\'re operating on exist?')
NO_URLS_MATCHED_TARGET = NO_URLS_MATCHED_PREFIX + ': %s'
class AbortException(StandardError):
class CommandException(StandardError):
class ControlCException(Exception):
class GcloudStorageTranslationError(Exception):
class HashMismatchException(Exception):
class IamChOnResourceWithConditionsException(Exception):
class InvalidUrlError(Exception):
class ExternalBinaryError(Exception):
  def __init__(self, reason):
  def __repr__(self):
  def __str__(self):
  def __init__(self, reason, informational=False):
  def __repr__(self):
  def __str__(self):
  def __init__(self, message):
  def __repr__(self):
  def __str__(self):
  def __init__(self, message):
  def __repr__(self):
  def __str__(self):
  def __init__(self, message):
  def __repr__(self): DEFAULT_SLICED_OBJECT_DOWNLOAD_THRESHOLD = '150M' class InvalidUrlError(Exception):
  """Exception raised when URL is invalid."""

  def __init__(self, message):
    Exception.__init__(self, message)
    self.message = message

  def __repr__(self):
    return str(self)

  def __str__(self):
    return 'InvalidUrlError: %s' % self.message class GcsJsonApi(CloudApi):
  """Google Cloud Storage JSON implementation of gsutil Cloud API."""

  def __init__(self,
               bucket_storage_uri_class,
               logger,
               status_queue,
               provider=None,
               credentials=None,
               debug=0,
               trace_token=None,
               perf_trace_token=None,
               user_project=None):
    """Performs necessary setup for interacting with Google Cloud Storage.

    Args:
      bucket_storage_uri_class: Unused.
      logger: logging.logger for outputting log messages.
      status_queue: Queue for relaying status to UI.
      provider: Unused.  This implementation supports only Google Cloud Storage.
      credentials: Credentials to be used for interacting with Google Cloud
                   Storage.
      debug: Debug level for the API implementation (0..3).
      trace_token: Trace token to pass to the API implementation.
      perf_trace_token: Performance trace token to use when making API calls.
      user_project: Project to be billed for this request.
    """
    # TODO: Plumb host_header for perfdiag / test_perfdiag.
    # TODO: Add jitter to apitools' http_wrapper retry mechanism.
    super(GcsJsonApi, self).__init__(bucket_storage_uri_class,
                                     logger,
                                     status_queue,
                                     provider='gs',
                                     debug=debug,
                                     trace_token=trace_token,
                                     perf_trace_token=perf_trace_token,
                                     user_project=user_project)

    self.certs_file = GetCertsFile()
    self.http = GetNewHttp()
    SetUpJsonCredentialsAndCache(self, logger, credentials=credentials)

    # Re-use download and upload connections. This class is only called
    # sequentially, but we can share TCP warmed-up connections across calls.
    self.download_http = self._GetNewDownloadHttp()
    self.upload_http = self._GetNewUploadHttp()
    if self.credentials:
      self.authorized_download_http = self.credentials.authorize(
          self.download_http)
      self.authorized_upload_http = self.credentials.authorize(self.upload_http)
    else:
      self.authorized_download_http = self.download_http
      self.authorized_upload_http = self.upload_http
    WrapDownloadHttpRequest(self.authorized_download_http)
    WrapUploadHttpRequest(self.authorized_upload_http)

    self.http_base = 'https://'
    gs_json_host = config.get('Credentials', 'gs_json_host', None)
    if (context_config.get_context_config() and
        context_config.get_context_config().use_client_certificate):
      if gs_json_host:
        raise ArgumentException(
            '"use_client_certificate" is enabled, which sets gsutil to use the'
            ' host {}. However, a custom host was set using'
            ' "gs_json_host": {}. Please set "use_client_certificate" to'
            ' "False" or comment out the "gs_json_host" line in the Boto'
            ' config.'.format(MTLS_HOST, gs_json_host))
      self.host_base = MTLS_HOST
    else:
      self.host_base = gs_json_host or DEFAULT_HOST

    if not gs_json_host:
      gs_host = config.get('Credentials', 'gs_host', None)
      if gs_host:
        raise ArgumentException(
            'JSON API is selected but gs_json_host is not configured, '
            'while gs_host is configured to %s. Please also configure '
            'gs_json_host and gs_json_port to match your desired endpoint.' %
            gs_host)

    gs_json_port = config.get('Credentials', 'gs_json_port', None)

    if not gs_json_port:
      gs_port = config.get('Credentials', 'gs_port', None)
      if gs_port:
        raise ArgumentException(
            'JSON API is selected but gs_json_port is not configured, '
            'while gs_port is configured to %s. Please also configure '
            'gs_json_host and gs_json_port to match your desired endpoint.' %
            gs_port)
      self.host_port = ''
    else:
      self.host_port = ':' + config.get('Credentials', 'gs_json_port')

    self.api_version = GetGcsJsonApiVersion()
    self.url_base = (self.http_base + self.host_base + self.host_port + '/' +
                     'storage/' + self.api_version + '/')

    self.global_params = apitools_messages.StandardQueryParameters(
        trace='token:%s' % trace_token) if trace_token else None

    additional_http_headers = {}

    gs_json_host_header = config.get('Credentials', 'gs_json_host_header', None)
    if gs_json_host_header and gs_json_host:
      additional_http_headers['Host'] = gs_json_host_header

    self._AddPerfTraceTokenToHeaders(additional_http_headers)
    self._AddReasonToHeaders(additional_http_headers)

    log_request = (debug >= 3)
    log_response = (debug >= 3)

    self.api_client = apitools_client.StorageV1(
        url=self.url_base,
        http=self.http,
        log_request=log_request,
        log_response=log_response,
        credentials=self.credentials,
        version=self.api_version,
        default_global_params=self.global_params,
        additional_http_headers=additional_http_headers)

    self.max_retry_wait = GetMaxRetryDelay()
    self.api_client.max_retry_wait = self.max_retry_wait

    self.num_retries = GetNumRetries()
    self.api_client.num_retries = self.num_retries

    self.api_client.retry_func = LogAndHandleRetries(
        status_queue=self.status_queue)
    self.api_client.overwrite_transfer_urls_with_client_base = True

    if isinstance(self.credentials, NoOpCredentials):
      # This API key is not secret and is used to identify gsutil during
      # anonymous requests.
      self.api_client.AddGlobalParam('key',
                                     'AIzaSyDnacJHrKma0048b13sh8cgxNUwulubmJM')

  def GetServiceAccountId(self):
    """Returns the service account email id."""
    if isinstance(self.credentials, ImpersonationCredentials):
      return self.credentials.service_account_id
    elif isinstance(self.credentials, ServiceAccountCredentials):
      return self.credentials.service_account_email
    else:
      raise CommandException(
          'Cannot get service account email id for the given '
          'credential type.')

  def _AddPerfTraceTokenToHeaders(self, headers):
    if self.perf_trace_token:
      headers['cookie'] = self.perf_trace_token

  def _AddReasonToHeaders(self, headers):
    request_reason = os.environ.get(REQUEST_REASON_ENV_VAR)
    if request_reason:
      headers[REQUEST_REASON_HEADER_KEY] = request_reason

  def _GetNewDownloadHttp(self):
    return GetNewHttp(http_class=HttpWithDownloadStream)

  def _GetNewUploadHttp(self):
    """Returns an upload-safe Http object (by disabling httplib2 retries)."""
    return GetNewHttp(http_class=HttpWithNoRetries)

  def _GetSignedContent(self, string_to_sign):
    """Returns the Signed Content."""
    if isinstance(self.credentials, ImpersonationCredentials):
      iam_cred_api = self.credentials.api
      service_account_id = self.credentials.service_account_id
      response = iam_cred_api.SignBlob(service_account_id, string_to_sign)
      return response.keyId, response.signedBlob
    elif isinstance(self.credentials, ServiceAccountCredentials):
      return self.credentials.sign_blob(string_to_sign)
    else:
      raise CommandException(
          'Authentication using a service account is required for signing '
          'the content.')

  def _FieldsContainsAclField(self, fields=None):
    """Checks Returns true if ACL related values are in fields set.

    Args:
      fields: list or set of fields. May be in GET ['acl'] or List
          ['items/acl'] call format.

    Returns:
      True if an ACL value is requested in the input fields, False otherwise.
    """
    return fields is None or _ACL_FIELDS_SET.intersection(set(fields))

  def _ValidateEncryptionFieldsForMetadataGetRequest(self, fields=None):
    """Ensures customerEncryption field is included if hashes are requested."""
    # In the context of fetching object metadata, we only need to check for
    # customerEncryption, the field present if an object is CSEK-encrypted. Note
    # that if the object is KMS-encrypted, no field checks are necessary; a
    # metadata get request will be able to access the object's stored kmsKeyName
    # field to access the key needed to decrypt hash fields).
    if (fields and _ENCRYPTED_HASHES_SET.intersection(set(fields)) and
        'customerEncryption' not in fields):
      raise ArgumentException(
          'gsutil client error: customerEncryption must be included when '
          'requesting potentially encrypted fields %s' % _ENCRYPTED_HASHES_SET)

  def GetBucketIamPolicy(self, bucket_name, provider=None, fields=None):
    apitools_request = apitools_messages.StorageBucketsGetIamPolicyRequest(
        bucket=bucket_name,
        userProject=self.user_project,
        optionsRequestedPolicyVersion=IAM_POLICY_VERSION)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))
    try:
      return self.api_client.buckets.GetIamPolicy(apitools_request,
                                                  global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def GetObjectIamPolicy(self,
                         bucket_name,
                         object_name,
                         generation=None,
                         provider=None,
                         fields=None):
    if generation is not None:
      generation = long(generation)
    apitools_request = apitools_messages.StorageObjectsGetIamPolicyRequest(
        bucket=bucket_name,
        object=object_name,
        generation=generation,
        userProject=self.user_project)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))
    try:
      return self.api_client.objects.GetIamPolicy(apitools_request,
                                                  global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e,
                                       bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  def SetObjectIamPolicy(self,
                         bucket_name,
                         object_name,
                         policy,
                         generation=None,
                         provider=None):
    if generation is not None:
      generation = long(generation)
    api_request = apitools_messages.StorageObjectsSetIamPolicyRequest(
        bucket=bucket_name,
        object=object_name,
        generation=generation,
        policy=policy,
        userProject=self.user_project)
    global_params = apitools_messages.StandardQueryParameters()
    try:
      return self.api_client.objects.SetIamPolicy(api_request,
                                                  global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e,
                                       bucket_name=bucket_name,
                                       object_name=object_name)

  def SetBucketIamPolicy(self, bucket_name, policy, provider=None):
    apitools_request = apitools_messages.StorageBucketsSetIamPolicyRequest(
        bucket=bucket_name, policy=policy, userProject=self.user_project)
    global_params = apitools_messages.StandardQueryParameters()
    try:
      return self.api_client.buckets.SetIamPolicy(apitools_request,
                                                  global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def SignUrl(self, method, duration, path, generation, logger, region,
              signed_headers, string_to_sign_debug):
    """See CloudApi class for function doc strings."""
    service_account_id = self.GetServiceAccountId()
    string_to_sign, canonical_query_string = CreatePayload(
        client_id=service_account_id,
        method=method,
        duration=duration,
        path=path,
        generation=generation,
        logger=logger,
        region=region,
        signed_headers=signed_headers,
        string_to_sign_debug=string_to_sign_debug)

    if six.PY3:
      string_to_sign = string_to_sign.encode(UTF8)

    key_id, raw_signature = self._GetSignedContent(string_to_sign)
    logger.debug('Key ID used to sign blob for service account "%s": "%s"' %
                 (service_account_id, key_id))
    return GetFinalUrl(raw_signature, signed_headers['host'], path,
                       canonical_query_string)

  def GetBucket(self, bucket_name, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageBucketsGetRequest.
                  ProjectionValueValuesEnum.noAcl)
    if self._FieldsContainsAclField(fields):
      projection = (apitools_messages.StorageBucketsGetRequest.
                    ProjectionValueValuesEnum.full)
    apitools_request = apitools_messages.StorageBucketsGetRequest(
        bucket=bucket_name,
        projection=projection,
        userProject=self.user_project)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))

    # Here and in list buckets, we have no way of knowing
    # whether we requested a field and didn't get it because it didn't exist
    # or because we didn't have permission to access it.
    try:
      return self.api_client.buckets.Get(apitools_request,
                                         global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def ListBucketAccessControls(self, bucket_name):
    """See CloudApi class for function doc strings."""
    try:
      request = apitools_messages.StorageBucketAccessControlsListRequest(
          bucket=bucket_name)
      return self.api_client.bucketAccessControls.List(request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def ListObjectAccessControls(self, bucket_name, object_name):
    """See CloudApi class for function doc strings."""
    try:
      request = apitools_messages.StorageObjectAccessControlsListRequest(
          bucket=bucket_name, object=object_name)
      return self.api_client.objectAccessControls.List(request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e,
                                       bucket_name=bucket_name,
                                       object_name=object_name)

  def PatchBucket(self,
                  bucket_name,
                  metadata,
                  canned_acl=None,
                  canned_def_acl=None,
                  preconditions=None,
                  provider=None,
                  fields=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageBucketsPatchRequest.
                  ProjectionValueValuesEnum.noAcl)
    if self._FieldsContainsAclField(fields):
      projection = (apitools_messages.StorageBucketsPatchRequest.
                    ProjectionValueValuesEnum.full)
    bucket_metadata = metadata

    if not preconditions:
      preconditions = Preconditions()

    # For blank metadata objects, we need to explicitly call
    # them out to apitools so it will send/erase them.
    apitools_include_fields = []
    for metadata_field in (
        'billing',
        'encryption',
        'lifecycle',
        'logging',
        'metadata',
        'retentionPolicy',
        'versioning',
        'website',
    ):
      attr = getattr(bucket_metadata, metadata_field, None)
      if attr and not encoding.MessageToDict(attr):
        setattr(bucket_metadata, metadata_field, None)
        apitools_include_fields.append(metadata_field)

    if bucket_metadata.cors and bucket_metadata.cors == REMOVE_CORS_CONFIG:
      bucket_metadata.cors = []
      apitools_include_fields.append('cors')

    if (bucket_metadata.defaultObjectAcl and
        bucket_metadata.defaultObjectAcl[0] == PRIVATE_DEFAULT_OBJ_ACL):
      bucket_metadata.defaultObjectAcl = []
      apitools_include_fields.append('defaultObjectAcl')

    predefined_acl = None
    if canned_acl:
      # Must null out existing ACLs to apply a canned ACL.
      apitools_include_fields.append('acl')
      predefined_acl = (apitools_messages.StorageBucketsPatchRequest.
                        PredefinedAclValueValuesEnum(
                            self._BucketCannedAclToPredefinedAcl(canned_acl)))

    predefined_def_acl = None
    if canned_def_acl:
      # Must null out existing default object ACLs to apply a canned ACL.
      apitools_include_fields.append('defaultObjectAcl')
      predefined_def_acl = (
          apitools_messages.StorageBucketsPatchRequest.
          PredefinedDefaultObjectAclValueValuesEnum(
              self._ObjectCannedAclToPredefinedAcl(canned_def_acl)))

    apitools_request = apitools_messages.StorageBucketsPatchRequest(
        bucket=bucket_name,
        bucketResource=bucket_metadata,
        projection=projection,
        ifMetagenerationMatch=preconditions.meta_gen_match,
        predefinedAcl=predefined_acl,
        predefinedDefaultObjectAcl=predefined_def_acl,
        userProject=self.user_project)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))
    with self.api_client.IncludeFields(apitools_include_fields):
      try:
        return self.api_client.buckets.Patch(apitools_request,
                                             global_params=global_params)
      except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
        self._TranslateExceptionAndRaise(e)

  def LockRetentionPolicy(self, bucket_name, metageneration, provider=None):
    try:
      metageneration = long(metageneration)
    except ValueError:
      raise ArgumentException(
          'LockRetentionPolicy Metageneration must be an integer.')

    apitools_request = (
        apitools_messages.StorageBucketsLockRetentionPolicyRequest(
            bucket=bucket_name,
            ifMetagenerationMatch=metageneration,
            userProject=self.user_project))
    global_params = apitools_messages.StandardQueryParameters()
    try:
      return self.api_client.buckets.LockRetentionPolicy(
          apitools_request, global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def CreateBucket(self,
                   bucket_name,
                   project_id=None,
                   metadata=None,
                   provider=None,
                   fields=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageBucketsInsertRequest.
                  ProjectionValueValuesEnum.noAcl)
    if self._FieldsContainsAclField(fields):
      projection = (apitools_messages.StorageBucketsInsertRequest.
                    ProjectionValueValuesEnum.full)
    if not metadata:
      metadata = apitools_messages.Bucket()
    metadata.name = bucket_name

    if metadata.location:
      metadata.location = metadata.location.lower()
    if metadata.storageClass:
      metadata.storageClass = metadata.storageClass.lower()

    project_id = PopulateProjectId(project_id)

    apitools_request = apitools_messages.StorageBucketsInsertRequest(
        bucket=metadata,
        project=project_id,
        projection=projection,
        userProject=self.user_project)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))
    try:
      return self.api_client.buckets.Insert(apitools_request,
                                            global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def DeleteBucket(self, bucket_name, preconditions=None, provider=None):
    """See CloudApi class for function doc strings."""
    if not preconditions:
      preconditions = Preconditions()

    apitools_request = apitools_messages.StorageBucketsDeleteRequest(
        bucket=bucket_name,
        ifMetagenerationMatch=preconditions.meta_gen_match,
        userProject=self.user_project)

    try:
      self.api_client.buckets.Delete(apitools_request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      if isinstance(
          self._TranslateApitoolsException(e, bucket_name=bucket_name),
          NotEmptyException):
        # If bucket is not empty, check to see if versioning is enabled and
        # signal that in the exception if it is.
        bucket_metadata = self.GetBucket(bucket_name, fields=['versioning'])
        if bucket_metadata.versioning and bucket_metadata.versioning.enabled:
          raise NotEmptyException('VersionedBucketNotEmpty',
                                  status=e.status_code)
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def ListBuckets(self, project_id=None, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageBucketsListRequest.
                  ProjectionValueValuesEnum.noAcl)
    if self._FieldsContainsAclField(fields):
      projection = (apitools_messages.StorageBucketsListRequest.
                    ProjectionValueValuesEnum.full)
    project_id = PopulateProjectId(project_id)

    apitools_request = apitools_messages.StorageBucketsListRequest(
        project=project_id,
        maxResults=NUM_BUCKETS_PER_LIST_PAGE,
        projection=projection,
        userProject=self.user_project)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      if 'nextPageToken' not in fields:
        fields.add('nextPageToken')
      global_params.fields = ','.join(set(fields))
    try:
      bucket_list = self.api_client.buckets.List(apitools_request,
                                                 global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e)

    for bucket in self._YieldBuckets(bucket_list):
      yield bucket

    while bucket_list.nextPageToken:
      apitools_request = apitools_messages.StorageBucketsListRequest(
          project=project_id,
          pageToken=bucket_list.nextPageToken,
          maxResults=NUM_BUCKETS_PER_LIST_PAGE,
          projection=projection,
          userProject=self.user_project)
      try:
        bucket_list = self.api_client.buckets.List(apitools_request,
                                                   global_params=global_params)
      except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
        self._TranslateExceptionAndRaise(e)

      for bucket in self._YieldBuckets(bucket_list):
        yield bucket

  def _YieldBuckets(self, bucket_list):
    """Yields buckets from a list returned by apitools."""
    if bucket_list.items:
      for bucket in bucket_list.items:
        yield bucket

  def ListObjects(self,
                  bucket_name,
                  prefix=None,
                  delimiter=None,
                  all_versions=None,
                  provider=None,
                  fields=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageObjectsListRequest.
                  ProjectionValueValuesEnum.noAcl)
    if self._FieldsContainsAclField(fields):
      projection = (apitools_messages.StorageObjectsListRequest.
                    ProjectionValueValuesEnum.full)
    apitools_request = apitools_messages.StorageObjectsListRequest(
        bucket=bucket_name,
        prefix=prefix,
        delimiter=delimiter,
        versions=all_versions,
        projection=projection,
        maxResults=NUM_OBJECTS_PER_LIST_PAGE,
        userProject=self.user_project)
    global_params = apitools_messages.StandardQueryParameters()

    if fields:
      fields = set(fields)
      if 'nextPageToken' not in fields:
        fields.add('nextPageToken')
      if ListToGetFields(
          list_fields=fields).intersection(_ENCRYPTED_HASHES_SET):
        # We may need to make a follow-up request to decrypt the hashes,
        # so ensure the required fields are present.
        fields.add('items/customerEncryption')
        fields.add('items/kmsKeyName')
        fields.add('items/generation')
        fields.add('items/name')
      global_params.fields = ','.join(fields)

    list_page = True
    next_page_token = None

    while list_page:
      list_page = False
      apitools_request = apitools_messages.StorageObjectsListRequest(
          bucket=bucket_name,
          prefix=prefix,
          delimiter=delimiter,
          versions=all_versions,
          projection=projection,
          pageToken=next_page_token,
          maxResults=NUM_OBJECTS_PER_LIST_PAGE,
          userProject=self.user_project)
      try:
        object_list = self.api_client.objects.List(apitools_request,
                                                   global_params=global_params)
      except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
        self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

      if object_list.nextPageToken:
        list_page = True
        next_page_token = object_list.nextPageToken

      for object_or_prefix in self._YieldObjectsAndPrefixes(object_list):
        if object_or_prefix.datatype == CloudApi.CsObjectOrPrefixType.OBJECT:
          decrypted_metadata = self._DecryptHashesIfPossible(
              bucket_name,
              object_or_prefix.data,
              fields=ListToGetFields(fields))
          if decrypted_metadata == _SKIP_LISTING_OBJECT:
            continue
          elif decrypted_metadata:
            object_or_prefix.data = decrypted_metadata
          else:
            # Object metadata was unencrypted or we did not have a matching
            # key; yield what we received in the initial listing.
            pass

        yield object_or_prefix

  def _DecryptHashesIfPossible(self, bucket_name, object_metadata, fields=None):
    """Attempts to decrypt object metadata.

    Helps fetch encrypted object metadata fields (if requested) for API method
    calls that requested them but did not receive them; this occurs for
    ListObjects(), along with GetObjectMetadata() calls for which the object is
    CSEK-encrypted but no decryption information was supplied.

    This method issues a StorageObjectsGetRequest (if object hash fields were
    requested but not retrieved), formatted appropriately depending on the
    encryption method.

    Args:
      bucket_name: String of bucket containing the object.
      object_metadata: apitools Object describing the object. Must include
          name, generation, and customerEncryption fields.
      fields: ListObjects-format fields to return.

    Returns:
      - apitools_messages.Object with decrypted hash fields if decryption was
      performed.
      - None if the object was not encrypted, or CSEK-encrypted but we could not
      find a matching decryption key.
      - gcs_json_api._SKIP_LISTING_OBJECT if the object no longer exists.
    """
    get_fields = ListToGetFields(list_fields=fields)

    if not self._ObjectIsEncrypted(object_metadata):
      self._RaiseIfNoDesiredHashesPresentForUnencryptedObject(object_metadata,
                                                              fields=get_fields)
      return None

    # If we reach this point, we know the object is encrypted.
    get_metadata_func = None
    if self._ObjectKMSEncryptedAndNeedHashes(object_metadata,
                                             fields=get_fields):
      # For CMK-encrypted objects, we can just call GetObjectMetadata.
      get_metadata_func = functools.partial(
          self._GetObjectMetadataHelper,
          bucket_name,
          object_metadata.name,
          generation=object_metadata.generation,
          fields=get_fields)
    elif self._ObjectCSEKEncryptedAndNeedHashes(object_metadata,
                                                fields=get_fields):
      # For CSEK-encrypted objects, we can do almost the same thing as above,
      # but we also need to add some info about the decryption key (if it's
      # available) into the request.
      key_sha256 = object_metadata.customerEncryption.keySha256
      if six.PY3:
        if not isinstance(key_sha256, bytes):
          key_sha256 = key_sha256.encode('ascii')
      decryption_key = FindMatchingCSEKInBotoConfig(key_sha256, config)
      if decryption_key:
        get_metadata_func = functools.partial(
            self._GetObjectMetadataHelper,
            bucket_name,
            object_metadata.name,
            generation=object_metadata.generation,
            fields=get_fields,
            decryption_tuple=CryptoKeyWrapperFromKey(decryption_key))

    if get_metadata_func is None:
      return

    try:
      return get_metadata_func()
    except NotFoundException:
      # If the object was deleted between the listing request and
      # the get request, do not include it in the list.
      return _SKIP_LISTING_OBJECT

  def _YieldObjectsAndPrefixes(self, object_list):
    # ls depends on iterating objects before prefixes for proper display.
    if object_list.items:
      for cloud_obj in object_list.items:
        yield CloudApi.CsObjectOrPrefix(cloud_obj,
                                        CloudApi.CsObjectOrPrefixType.OBJECT)
    if object_list.prefixes:
      for prefix in object_list.prefixes:
        yield CloudApi.CsObjectOrPrefix(prefix,
                                        CloudApi.CsObjectOrPrefixType.PREFIX)

  @contextmanager
  def _ApitoolsRequestHeaders(self, headers):
    """Wrapped code makes apitools requests with the specified headers."""
    if headers:
      old_headers = self.api_client.additional_http_headers.copy()
      self.api_client.additional_http_headers.update(headers)
    yield
    if headers:
      self.api_client.additional_http_headers = old_headers

  def _EncryptionHeadersFromTuple(self, crypto_tuple=None):
    """Returns headers dict matching crypto_tuple, or empty dict."""
    headers = {}
    if crypto_tuple and crypto_tuple.crypto_type == CryptoKeyType.CSEK:
      headers['x-goog-encryption-algorithm'] = crypto_tuple.crypto_alg
      headers['x-goog-encryption-key'] = crypto_tuple.crypto_key
      headers['x-goog-encryption-key-sha256'] = (
          Base64Sha256FromBase64EncryptionKey(crypto_tuple.crypto_key))
    return headers

  def _RewriteCryptoHeadersFromTuples(self,
                                      decryption_tuple=None,
                                      encryption_tuple=None):
    """Returns headers dict matching the provided tuples, or empty dict."""
    headers = {}
    if decryption_tuple and decryption_tuple.crypto_type == CryptoKeyType.CSEK:
      headers['x-goog-copy-source-encryption-algorithm'] = (
          decryption_tuple.crypto_alg)
      headers['x-goog-copy-source-encryption-key'] = (
          decryption_tuple.crypto_key)
      headers['x-goog-copy-source-encryption-key-sha256'] = (
          decryption_tuple.crypto_key_sha256)

    if encryption_tuple and encryption_tuple.crypto_type == CryptoKeyType.CSEK:
      headers['x-goog-encryption-algorithm'] = encryption_tuple.crypto_alg
      headers['x-goog-encryption-key'] = encryption_tuple.crypto_key
      headers['x-goog-encryption-key-sha256'] = (
          encryption_tuple.crypto_key_sha256)

    return headers

  def _GetObjectMetadataHelper(self,
                               bucket_name,
                               object_name,
                               generation=None,
                               fields=None,
                               decryption_tuple=None):
    """Get metadata, adding necessary headers for CSEK decryption if necessary.

    Args:
      bucket_name: (str) Bucket containing the object.
      object_name: (str) Object name.
      generation: (int) Generation of the object to retrieve.
      fields: (list<str>) If present, return only these Object metadata fields,
          for example: ['acl', 'updated'].
      decryption_tuple: (gslib.utils.encryption_helper.CryptoKeyWrapper) The
          CryptoKeyWrapper for the desired decryption key.

    Raises:
      ArgumentException for errors during input validation.
      ServiceException for errors interacting with cloud storage providers.

    Returns:
      (apitools_message.Object) The retrieved object metadata.
    """
    apitools_request = self._CreateApitoolsObjectMetadataGetRequest(
        bucket_name, object_name, generation=generation, fields=fields)
    global_params = self._GetApitoolsObjectMetadataGlobalParams(fields=fields)

    try:
      # Add special headers if fetching metadata for a CSEK-encrypted object.
      if decryption_tuple:
        with self._ApitoolsRequestHeaders(
            self._EncryptionHeadersFromTuple(crypto_tuple=decryption_tuple)):
          return self.api_client.objects.Get(apitools_request,
                                             global_params=global_params)
      else:
        return self.api_client.objects.Get(apitools_request,
                                           global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e,
                                       bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  def _CreateApitoolsObjectMetadataGetRequest(self,
                                              bucket_name,
                                              object_name,
                                              generation=None,
                                              fields=None):
    self._ValidateEncryptionFieldsForMetadataGetRequest(fields=fields)
    projection = (apitools_messages.StorageObjectsGetRequest.
                  ProjectionValueValuesEnum.noAcl)
    if self._FieldsContainsAclField(fields):
      projection = (apitools_messages.StorageObjectsGetRequest.
                    ProjectionValueValuesEnum.full)
    if generation:
      generation = long(generation)

    return apitools_messages.StorageObjectsGetRequest(
        bucket=bucket_name,
        object=object_name,
        projection=projection,
        generation=generation,
        userProject=self.user_project)

  def _GetApitoolsObjectMetadataGlobalParams(self, fields=None):
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))
    return global_params

  def GetObjectMetadata(self,
                        bucket_name,
                        object_name,
                        generation=None,
                        provider=None,
                        fields=None):
    """See CloudApi class for function doc strings."""
    try:
      object_metadata = self._GetObjectMetadataHelper(bucket_name,
                                                      object_name,
                                                      generation=generation,
                                                      fields=fields)
      if not self._ObjectCSEKEncryptedAndNeedHashes(object_metadata,
                                                    fields=fields):
        return object_metadata

      # CSEK-encrypted objects require corresponding decryption information to
      # be added to the StorageObjectsGetRequest in order to retrieve hash
      # fields.
      key_sha256 = object_metadata.customerEncryption.keySha256
      if six.PY3:
        if not isinstance(key_sha256, bytes):
          key_sha256 = key_sha256.encode('ascii')
      decryption_key = FindMatchingCSEKInBotoConfig(key_sha256, config)
      if not decryption_key:
        raise EncryptionException(
            'Missing decryption key with SHA256 hash %s. No decryption key '
            'matches object %s://%s/%s' %
            (key_sha256, self.provider, bucket_name, object_name))
      if six.PY3:
        if not isinstance(decryption_key, bytes):
          decryption_key = decryption_key.encode(UTF8)
      return self._GetObjectMetadataHelper(
          bucket_name,
          object_name,
          generation=generation,
          fields=fields,
          decryption_tuple=CryptoKeyWrapperFromKey(decryption_key))
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e,
                                       bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  def _ObjectIsEncrypted(self, object_metadata):
    return bool(object_metadata.customerEncryption or
                object_metadata.kmsKeyName)

  def _RaiseIfNoDesiredHashesPresentForUnencryptedObject(
      self, object_metadata, fields=None):
    """Raise an exception if none of the desired hash fields are present."""
    if fields is None:
      return

    md5_requested_but_missing = ('md5Hash' in fields and
                                 not object_metadata.md5Hash)
    crc32c_requested_but_missing = ('crc32c' in fields and
                                    not object_metadata.crc32c)

    # md5 not available for composite objects; in this case, don't raise an
    # exception if that was the only hash we requested.
    if (md5_requested_but_missing and not object_metadata.componentCount and
        crc32c_requested_but_missing):
      raise ServiceException(
          'Service did not provide requested hashes in metadata for '
          'unencrypted object %s' % object_metadata.name)

  def _ObjectKMSEncryptedAndNeedHashes(self, object_metadata, fields=None):
    """Checks if an object is KMS-encrypted and lacking requested hash fields.

    Args:
      object_metadata: (apitools_messages.Object) Object metadata returned from
          an API call; may or may not contain hash-related metadata fields.
      fields: (list<str>) Requested object metadata fields.

    Returns:
      True if a GetObjectMetadata request should be issued on the KMS-encrypted
      object in order to retrieve encrypted hash fields, False otherwise.
    """
    if fields is not None and not _ENCRYPTED_HASHES_SET.intersection(fields):
      # No potentially encrypted metadata fields were requested.
      return False

    return (object_metadata.kmsKeyName and not object_metadata.crc32c and
            not object_metadata.md5Hash)

  def _ObjectCSEKEncryptedAndNeedHashes(self, object_metadata, fields=None):
    """Checks if an object is CSEK-encrypted and lacking requested hash fields.

    Args:
      object_metadata: (apitools_messages.Object) Object metadata returned from
          an API call; may or may not contain hash-related metadata fields.
      fields: (list<str>) Requested object metadata fields.

    Returns:
      True if a GetObjectMetadata request including a decryption key should be
      issued on the CSEK-encrypted object in order to retrieve encrypted hash
      fields, False otherwise.
    """
    if fields is None or not _ENCRYPTED_HASHES_SET.intersection(fields):
      # No potentially encrypted metadata fields were requested.
      return False

    return (object_metadata.customerEncryption and
            not object_metadata.crc32c and not object_metadata.md5Hash)

  def GetObjectMedia(self,
                     bucket_name,
                     object_name,
                     download_stream,
                     provider=None,
                     generation=None,
                     object_size=None,
                     compressed_encoding=False,
                     download_strategy=CloudApi.DownloadStrategy.ONE_SHOT,
                     start_byte=0,
                     end_byte=None,
                     progress_callback=None,
                     serialization_data=None,
                     digesters=None,
                     decryption_tuple=None):
    """See CloudApi class for function doc strings."""
    # This implementation will get the object metadata first if we don't pass it
    # in via serialization_data.
    if generation:
      generation = long(generation)

    # 'outer_total_size' is only used for formatting user output, and is
    # expected to be one higher than the last byte that should be downloaded.
    # TODO: Change DownloadCallbackConnectionClassFactory and progress callbacks
    # to more elegantly handle total size for components of files.
    outer_total_size = object_size
    if end_byte:
      outer_total_size = end_byte + 1
    elif serialization_data:
      outer_total_size = json.loads(
          six.ensure_str(serialization_data))['total_size']

    if progress_callback:
      if outer_total_size is None:
        raise ArgumentException('Download size is required when callbacks are '
                                'requested for a download, but no size was '
                                'provided.')
      progress_callback(start_byte, outer_total_size)

    bytes_downloaded_container = BytesTransferredContainer()
    bytes_downloaded_container.bytes_transferred = start_byte

    callback_class_factory = DownloadCallbackConnectionClassFactory(
        bytes_downloaded_container,
        total_size=outer_total_size,
        progress_callback=progress_callback,
        digesters=digesters)
    download_http_class = callback_class_factory.GetConnectionClass()

    # Point our download HTTP at our download stream.
    self.download_http.stream = download_stream
    self.download_http.connections = {'https': download_http_class}

    if serialization_data:
      # If we have an apiary trace token, add it to the URL.
      # TODO: Add query parameter support to apitools downloads so there is
      # a well-defined way to express query parameters. Currently, we assume
      # the URL ends in ?alt=media, and this will break if that changes.
      if self.trace_token:
        serialization_dict = json.loads(six.ensure_str(serialization_data))
        serialization_dict['url'] += '&trace=token%%3A%s' % self.trace_token
        serialization_data = json.dumps(serialization_dict)

      apitools_download = apitools_transfer.Download.FromData(
          download_stream,
          serialization_data,
          self.api_client.http,
          num_retries=self.num_retries,
          client=self.api_client)
    else:
      apitools_download = apitools_transfer.Download.FromStream(
          download_stream,
          auto_transfer=False,
          total_size=object_size,
          num_retries=self.num_retries)

    apitools_download.bytes_http = self.authorized_download_http
    apitools_request = apitools_messages.StorageObjectsGetRequest(
        bucket=bucket_name,
        object=object_name,
        generation=generation,
        userProject=self.user_project)

    # Disable retries in apitools. We will handle them explicitly for
    # resumable downloads; one-shot downloads are not retryable as we do
    # not track how many bytes were written to the stream.
    apitools_download.retry_func = LogAndHandleRetries(
        is_data_transfer=True, status_queue=self.status_queue)

    try:
      if download_strategy == CloudApi.DownloadStrategy.RESUMABLE:
        return self._PerformResumableDownload(
            bucket_name,
            object_name,
            download_stream,
            apitools_request,
            apitools_download,
            bytes_downloaded_container,
            compressed_encoding=compressed_encoding,
            generation=generation,
            start_byte=start_byte,
            end_byte=end_byte,
            serialization_data=serialization_data,
            decryption_tuple=decryption_tuple)
      else:
        return self._PerformDownload(bucket_name,
                                     object_name,
                                     download_stream,
                                     apitools_request,
                                     apitools_download,
                                     generation=generation,
                                     compressed_encoding=compressed_encoding,
                                     start_byte=start_byte,
                                     end_byte=end_byte,
                                     serialization_data=serialization_data,
                                     decryption_tuple=decryption_tuple)
    # If you are fighting a redacted exception spew in multiprocess/multithread
    # calls, add your exception to TRANSLATABLE_APITOOLS_EXCEPTIONS and put
    # something like this immediately after the following except statement:
    # import sys, traceback; sys.stderr.write('\n{}\n'.format(
    #     traceback.format_exc())); sys.stderr.flush()
    # This may hang, but you should get a stack trace spew after Ctrl-C.
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e,
                                       bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  def _PerformResumableDownload(self,
                                bucket_name,
                                object_name,
                                download_stream,
                                apitools_request,
                                apitools_download,
                                bytes_downloaded_container,
                                generation=None,
                                compressed_encoding=False,
                                start_byte=0,
                                end_byte=None,
                                serialization_data=None,
                                decryption_tuple=None):
    retries = 0
    last_progress_byte = start_byte
    while retries <= self.num_retries:
      try:
        return self._PerformDownload(bucket_name,
                                     object_name,
                                     download_stream,
                                     apitools_request,
                                     apitools_download,
                                     generation=generation,
                                     compressed_encoding=compressed_encoding,
                                     start_byte=start_byte,
                                     end_byte=end_byte,
                                     serialization_data=serialization_data,
                                     decryption_tuple=decryption_tuple)
      except HTTP_TRANSFER_EXCEPTIONS as e:
        self._ValidateHttpAccessTokenRefreshError(e)
        start_byte = download_stream.tell()
        bytes_downloaded_container.bytes_transferred = start_byte
        if start_byte > last_progress_byte:
          # We've made progress, so allow a fresh set of retries.
          last_progress_byte = start_byte
          retries = 0
        retries += 1
        if retries > self.num_retries:
          ##### DEBUG
          # If you are fighting a redacted exception spew in
          # multiprocess/multithread calls, add your exception to
          # HTTP_TRANSFER_EXCEPTIONS and uncomment the following block.  This
          # may hang, but you should get a stack trace spew after Ctrl-C.
          #####
          # import re, sys, traceback
          # from gslib.utils import text_util
          # message = 'some exception happened'
          # stack_trace = traceback.format_exc()
          # err = ('DEBUG: Exception stack trace:\n    %s\n%s\n' % (
          #     re.sub('\\n', '\n    ', stack_trace),
          #     message,
          # ))
          # dest_fd = sys.stderr
          # text_util.print_to_fd(err, end='', file=dest_fd)
          # dest_fd.flush()
          ##### END DEBUG
          raise ResumableDownloadException(
              'Transfer failed after %d retries. Final exception: %s' %
              (self.num_retries, GetPrintableExceptionString(e)))
        time.sleep(CalculateWaitForRetry(retries, max_wait=self.max_retry_wait))
        if self.logger.isEnabledFor(logging.DEBUG):
          self.logger.debug(
              'Retrying download from byte %s after exception: %s. Trace: %s',
              start_byte, GetPrintableExceptionString(e),
              traceback.format_exc())
        apitools_http_wrapper.RebuildHttpConnections(
            apitools_download.bytes_http)

  def _PerformDownload(self,
                       bucket_name,
                       object_name,
                       download_stream,
                       apitools_request,
                       apitools_download,
                       generation=None,
                       compressed_encoding=False,
                       start_byte=0,
                       end_byte=None,
                       serialization_data=None,
                       decryption_tuple=None):
    if not serialization_data:
      try:
        self.api_client.objects.Get(apitools_request,
                                    download=apitools_download)
      except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
        self._TranslateExceptionAndRaise(e,
                                         bucket_name=bucket_name,
                                         object_name=object_name,
                                         generation=generation)

    # Disable apitools' default print callbacks.
    def _NoOpCallback(unused_response, unused_download_object):
      pass

    # TODO: If we have a resumable download with accept-encoding:gzip
    # on a object that is compressible but not in gzip form in the cloud,
    # on-the-fly compression may gzip the object.  In this case if our
    # download breaks, future requests will ignore the range header and just
    # return the object (gzipped) in its entirety.  Ideally, we would unzip
    # the bytes that we have locally and send a range request without
    # accept-encoding:gzip so that we can download only the (uncompressed) bytes
    # that we don't yet have.

    # Since bytes_http is created in this function, we don't get the
    # user-agent header from api_client's http automatically.
    additional_headers = {
        'user-agent': self.api_client.user_agent,
    }
    AddAcceptEncodingGzipIfNeeded(additional_headers,
                                  compressed_encoding=compressed_encoding)

    self._AddPerfTraceTokenToHeaders(additional_headers)
    self._AddReasonToHeaders(additional_headers)
    additional_headers.update(
        self._EncryptionHeadersFromTuple(decryption_tuple))

    if start_byte or end_byte is not None:
      apitools_download.GetRange(additional_headers=additional_headers,
                                 start=start_byte,
                                 end=end_byte,
                                 use_chunks=False)
    else:
      apitools_download.StreamMedia(callback=_NoOpCallback,
                                    finish_callback=_NoOpCallback,
                                    additional_headers=additional_headers,
                                    use_chunks=False)
    return apitools_download.encoding

  def PatchObjectMetadata(self,
                          bucket_name,
                          object_name,
                          metadata,
                          canned_acl=None,
                          generation=None,
                          preconditions=None,
                          provider=None,
                          fields=None,
                          user_project=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageObjectsPatchRequest.
                  ProjectionValueValuesEnum.noAcl)
    if self._FieldsContainsAclField(fields):
      projection = (apitools_messages.StorageObjectsPatchRequest.
                    ProjectionValueValuesEnum.full)

    if not preconditions:
      preconditions = Preconditions()

    if generation:
      generation = long(generation)

    predefined_acl = None
    apitools_include_fields = []
    if canned_acl:
      # Must null out existing ACLs to apply a canned ACL.
      apitools_include_fields.append('acl')
      predefined_acl = (apitools_messages.StorageObjectsPatchRequest.
                        PredefinedAclValueValuesEnum(
                            self._ObjectCannedAclToPredefinedAcl(canned_acl)))

    # Provide the ability to delete response headers from metadata.
    if metadata.cacheControl == '':
      apitools_include_fields.append('cacheControl')
    if metadata.contentDisposition == '':
      apitools_include_fields.append('contentDisposition')
    if metadata.contentEncoding == '':
      apitools_include_fields.append('contentEncoding')
    if metadata.contentLanguage == '':
      apitools_include_fields.append('contentLanguage')

    apitools_request = apitools_messages.StorageObjectsPatchRequest(
        bucket=bucket_name,
        object=object_name,
        objectResource=metadata,
        generation=generation,
        projection=projection,
        ifGenerationMatch=preconditions.gen_match,
        ifMetagenerationMatch=preconditions.meta_gen_match,
        predefinedAcl=predefined_acl,
        userProject=self.user_project)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))

    try:
      with self.api_client.IncludeFields(apitools_include_fields):
        return self.api_client.objects.Patch(apitools_request,
                                             global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e,
                                       bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  def _UploadObject(self,
                    upload_stream,
                    object_metadata,
                    canned_acl=None,
                    size=None,
                    preconditions=None,
                    encryption_tuple=None,
                    provider=None,
                    fields=None,
                    serialization_data=None,
                    tracker_callback=None,
                    progress_callback=None,
                    apitools_strategy=apitools_transfer.SIMPLE_UPLOAD,
                    total_size=0,
                    gzip_encoded=False):
    # pylint: disable=g-doc-args
    """Upload implementation. Cloud API arguments, plus two more.

    Additional args:
      apitools_strategy: SIMPLE_UPLOAD or RESUMABLE_UPLOAD.
      total_size: Total size of the upload; None if it is unknown (streaming).

    Returns:
      Uploaded object metadata.
    """
    # pylint: enable=g-doc-args
    ValidateDstObjectMetadata(object_metadata)
    predefined_acl = None
    if canned_acl:
      predefined_acl = (apitools_messages.StorageObjectsInsertRequest.
                        PredefinedAclValueValuesEnum(
                            self._ObjectCannedAclToPredefinedAcl(canned_acl)))

    bytes_uploaded_container = BytesTransferredContainer()

    if progress_callback and size:
      total_size = size
      progress_callback(0, size)

    callback_class_factory = UploadCallbackConnectionClassFactory(
        bytes_uploaded_container,
        total_size=total_size,
        progress_callback=progress_callback,
        logger=self.logger,
        debug=self.debug)

    upload_http_class = callback_class_factory.GetConnectionClass()
    self.upload_http.connections = {
        'http': upload_http_class,
        'https': upload_http_class,
    }

    # Since bytes_http is created in this function, we don't get the
    # user-agent header from api_client's http automatically.
    additional_headers = {
        'user-agent': self.api_client.user_agent,
    }
    self._AddPerfTraceTokenToHeaders(additional_headers)
    self._AddReasonToHeaders(additional_headers)

    try:
      content_type = None
      apitools_request = None
      global_params = None

      if not serialization_data:
        # This is a new upload, set up initial upload state.
        content_type = object_metadata.contentType
        if not content_type:
          content_type = DEFAULT_CONTENT_TYPE

        if not preconditions:
          preconditions = Preconditions()

        apitools_request = apitools_messages.StorageObjectsInsertRequest(
            bucket=object_metadata.bucket,
            object=object_metadata,
            ifGenerationMatch=preconditions.gen_match,
            ifMetagenerationMatch=preconditions.meta_gen_match,
            predefinedAcl=predefined_acl,
            userProject=self.user_project)
        global_params = apitools_messages.StandardQueryParameters()
        if fields:
          global_params.fields = ','.join(set(fields))

      encryption_headers = self._EncryptionHeadersFromTuple(
          crypto_tuple=encryption_tuple)

      if apitools_strategy == apitools_transfer.SIMPLE_UPLOAD:
        # One-shot upload.
        apitools_upload = apitools_transfer.Upload(upload_stream,
                                                   content_type,
                                                   total_size=size,
                                                   auto_transfer=True,
                                                   num_retries=self.num_retries,
                                                   gzip_encoded=gzip_encoded)
        apitools_upload.strategy = apitools_strategy
        apitools_upload.bytes_http = self.authorized_upload_http

        with self._ApitoolsRequestHeaders(encryption_headers):
          return self.api_client.objects.Insert(apitools_request,
                                                upload=apitools_upload,
                                                global_params=global_params)
      else:  # Resumable upload.
        # Since bytes_http is created in this function, we don't get the
        # user-agent header from api_client's http automatically.
        additional_headers = {
            'user-agent': self.api_client.user_agent,
        }
        additional_headers.update(encryption_headers)
        self._AddPerfTraceTokenToHeaders(additional_headers)
        self._AddReasonToHeaders(additional_headers)

        return self._PerformResumableUpload(
            upload_stream, self.authorized_upload_http, content_type, size,
            serialization_data, apitools_strategy, apitools_request,
            global_params, bytes_uploaded_container, tracker_callback,
            additional_headers, progress_callback, gzip_encoded)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      not_found_exception = CreateNotFoundExceptionForObjectWrite(
          self.provider, object_metadata.bucket)
      self._TranslateExceptionAndRaise(e,
                                       bucket_name=object_metadata.bucket,
                                       object_name=object_metadata.name,
                                       not_found_exception=not_found_exception)

  def _PerformResumableUpload(self, upload_stream, authorized_upload_http,
                              content_type, size, serialization_data,
                              apitools_strategy, apitools_request,
                              global_params, bytes_uploaded_container,
                              tracker_callback, addl_headers, progress_callback,
                              gzip_encoded):
    try:
      if serialization_data:
        # Resuming an existing upload.
        apitools_upload = apitools_transfer.Upload.FromData(
            upload_stream,
            serialization_data,
            self.api_client.http,
            num_retries=self.num_retries,
            gzip_encoded=gzip_encoded,
            client=self.api_client)
        apitools_upload.chunksize = GetJsonResumableChunkSize()
        apitools_upload.bytes_http = authorized_upload_http
      else:
        # New resumable upload.
        apitools_upload = apitools_transfer.Upload(
            upload_stream,
            content_type,
            total_size=size,
            chunksize=GetJsonResumableChunkSize(),
            auto_transfer=False,
            num_retries=self.num_retries,
            gzip_encoded=gzip_encoded)
        apitools_upload.strategy = apitools_strategy
        apitools_upload.bytes_http = authorized_upload_http
        with self._ApitoolsRequestHeaders(addl_headers):
          self.api_client.objects.Insert(apitools_request,
                                         upload=apitools_upload,
                                         global_params=global_params)
      # Disable retries in apitools. We will handle them explicitly here.
      apitools_upload.retry_func = LogAndHandleRetries(
          is_data_transfer=True, status_queue=self.status_queue)

      # Disable apitools' default print callbacks.
      def _NoOpCallback(unused_response, unused_upload_object):
        pass

      # If we're resuming an upload, apitools has at this point received
      # from the server how many bytes it already has. Update our
      # callback class with this information.
      bytes_uploaded_container.bytes_transferred = apitools_upload.progress

      if tracker_callback:
        tracker_callback(json.dumps(apitools_upload.serialization_data))

      retries = 0
      last_progress_byte = apitools_upload.progress
      while retries <= self.num_retries:
        try:
          # TODO: On retry, this will seek to the bytes that the server has,
          # causing the hash to be recalculated. Make HashingFileUploadWrapper
          # save a digest according to json_resumable_chunk_size.
          if not gzip_encoded and size and not JsonResumableChunkSizeDefined():
            # If size is known and the request doesn't need to be compressed,
            # we can send it all in one request and avoid making a
            # round-trip per chunk. Compression is not supported for
            # non-chunked streaming uploads because supporting resumability
            # for that feature results in degraded upload performance and
            # adds significant complexity to the implementation.
            http_response = apitools_upload.StreamMedia(
                callback=_NoOpCallback,
                finish_callback=_NoOpCallback,
                additional_headers=addl_headers)
          else:
            # Otherwise it's a streaming request and we need to ensure that we
            # send the bytes in chunks so that we can guarantee that we never
            # need to seek backwards more than our buffer (and also that the
            # chunks are aligned to 256KB).
            http_response = apitools_upload.StreamInChunks(
                callback=_NoOpCallback,
                finish_callback=_NoOpCallback,
                additional_headers=addl_headers)
          processed_response = self.api_client.objects.ProcessHttpResponse(
              self.api_client.objects.GetMethodConfig('Insert'), http_response)
          if size is None and progress_callback:
            # Make final progress callback; total size should now be known.
            # This works around the fact the send function counts header bytes.
            # However, this will make the progress appear to go slightly
            # backwards at the end.
            progress_callback(apitools_upload.total_size,
                              apitools_upload.total_size)
          return processed_response
        except HTTP_TRANSFER_EXCEPTIONS as e:
          self._ValidateHttpAccessTokenRefreshError(e)
          apitools_http_wrapper.RebuildHttpConnections(
              apitools_upload.bytes_http)
          while retries <= self.num_retries:
            try:
              # TODO: Simulate the refresh case in tests. Right now, our
              # mocks are not complex enough to simulate a failure.
              apitools_upload.RefreshResumableUploadState()
              start_byte = apitools_upload.progress
              bytes_uploaded_container.bytes_transferred = start_byte
              break
            except HTTP_TRANSFER_EXCEPTIONS as e2:
              self._ValidateHttpAccessTokenRefreshError(e2)
              apitools_http_wrapper.RebuildHttpConnections(
                  apitools_upload.bytes_http)
              retries += 1
              if retries > self.num_retries:
                raise ResumableUploadException(
                    'Transfer failed after %d retries. Final exception: %s' %
                    (self.num_retries, e2))
              time.sleep(
                  CalculateWaitForRetry(retries, max_wait=self.max_retry_wait))
          if start_byte > last_progress_byte:
            # We've made progress, so allow a fresh set of retries.
            last_progress_byte = start_byte
            retries = 0
          else:
            retries += 1
            if retries > self.num_retries:
              raise ResumableUploadException(
                  'Transfer failed after %d retries. Final exception: %s' %
                  (self.num_retries, GetPrintableExceptionString(e)))
            time.sleep(
                CalculateWaitForRetry(retries, max_wait=self.max_retry_wait))
          if self.logger.isEnabledFor(logging.DEBUG):
            self.logger.debug(
                'Retrying upload from byte %s after exception: %s. Trace: %s',
                start_byte, GetPrintableExceptionString(e),
                traceback.format_exc())
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      resumable_ex = self._TranslateApitoolsResumableUploadException(e)
      if resumable_ex:
        raise resumable_ex
      else:
        raise

  def UploadObject(self,
                   upload_stream,
                   object_metadata,
                   canned_acl=None,
                   size=None,
                   preconditions=None,
                   progress_callback=None,
                   encryption_tuple=None,
                   provider=None,
                   fields=None,
                   gzip_encoded=False):
    """See CloudApi class for function doc strings."""
    return self._UploadObject(upload_stream,
                              object_metadata,
                              canned_acl=canned_acl,
                              size=size,
                              preconditions=preconditions,
                              progress_callback=progress_callback,
                              encryption_tuple=encryption_tuple,
                              fields=fields,
                              apitools_strategy=apitools_transfer.SIMPLE_UPLOAD,
                              gzip_encoded=gzip_encoded)

  def UploadObjectStreaming(self,
                            upload_stream,
                            object_metadata,
                            canned_acl=None,
                            preconditions=None,
                            progress_callback=None,
                            encryption_tuple=None,
                            provider=None,
                            fields=None,
                            gzip_encoded=False):
    """See CloudApi class for function doc strings."""
    # Streaming indicated by not passing a size.
    # Resumable capabilities are present up to the resumable chunk size using
    # a buffered stream.
    return self._UploadObject(
        upload_stream,
        object_metadata,
        canned_acl=canned_acl,
        preconditions=preconditions,
        progress_callback=progress_callback,
        encryption_tuple=encryption_tuple,
        fields=fields,
        apitools_strategy=apitools_transfer.RESUMABLE_UPLOAD,
        total_size=None,
        gzip_encoded=gzip_encoded)

  def UploadObjectResumable(self,
                            upload_stream,
                            object_metadata,
                            canned_acl=None,
                            preconditions=None,
                            size=None,
                            serialization_data=None,
                            tracker_callback=None,
                            progress_callback=None,
                            encryption_tuple=None,
                            provider=None,
                            fields=None,
                            gzip_encoded=False):
    """See CloudApi class for function doc strings."""
    return self._UploadObject(
        upload_stream,
        object_metadata,
        canned_acl=canned_acl,
        preconditions=preconditions,
        fields=fields,
        size=size,
        serialization_data=serialization_data,
        tracker_callback=tracker_callback,
        progress_callback=progress_callback,
        encryption_tuple=encryption_tuple,
        apitools_strategy=apitools_transfer.RESUMABLE_UPLOAD,
        gzip_encoded=gzip_encoded)

  def CopyObject(self,
                 src_obj_metadata,
                 dst_obj_metadata,
                 src_generation=None,
                 canned_acl=None,
                 preconditions=None,
                 progress_callback=None,
                 max_bytes_per_call=None,
                 encryption_tuple=None,
                 decryption_tuple=None,
                 provider=None,
                 fields=None):
    """See CloudApi class for function doc strings."""
    ValidateDstObjectMetadata(dst_obj_metadata)
    predefined_acl = None
    if canned_acl:
      predefined_acl = (apitools_messages.StorageObjectsRewriteRequest.
                        DestinationPredefinedAclValueValuesEnum(
                            self._ObjectCannedAclToPredefinedAcl(canned_acl)))

    if src_generation:
      src_generation = long(src_generation)

    if not preconditions:
      preconditions = Preconditions()

    projection = (apitools_messages.StorageObjectsRewriteRequest.
                  ProjectionValueValuesEnum.noAcl)
    if self._FieldsContainsAclField(fields):
      projection = (apitools_messages.StorageObjectsRewriteRequest.
                    ProjectionValueValuesEnum.full)
    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      # Rewrite returns the resultant object under the 'resource' field.
      new_fields = set([
          'done',
          'objectSize',
          'rewriteToken',
          'totalBytesRewritten',
      ])
      for field in fields:
        new_fields.add('resource/' + field)
      global_params.fields = ','.join(set(new_fields))

    dec_key_sha256 = None
    if decryption_tuple and decryption_tuple.crypto_type == CryptoKeyType.CSEK:
      dec_key_sha256 = decryption_tuple.crypto_key_sha256

    enc_key_sha256 = None
    if encryption_tuple:
      if encryption_tuple.crypto_type == CryptoKeyType.CSEK:
        enc_key_sha256 = encryption_tuple.crypto_key_sha256
      elif encryption_tuple.crypto_type == CryptoKeyType.CMEK:
        dst_obj_metadata.kmsKeyName = encryption_tuple.crypto_key

    # Check to see if we are resuming a rewrite.
    tracker_file_name = GetRewriteTrackerFilePath(src_obj_metadata.bucket,
                                                  src_obj_metadata.name,
                                                  dst_obj_metadata.bucket,
                                                  dst_obj_metadata.name, 'JSON')
    rewrite_params_hash = HashRewriteParameters(
        src_obj_metadata,
        dst_obj_metadata,
        projection,
        src_generation=src_generation,
        gen_match=preconditions.gen_match,
        meta_gen_match=preconditions.meta_gen_match,
        canned_acl=predefined_acl,
        max_bytes_per_call=max_bytes_per_call,
        src_dec_key_sha256=dec_key_sha256,
        dst_enc_key_sha256=enc_key_sha256,
        fields=global_params.fields)
    resume_rewrite_token = ReadRewriteTrackerFile(tracker_file_name,
                                                  rewrite_params_hash)
    crypto_headers = self._RewriteCryptoHeadersFromTuples(
        decryption_tuple=decryption_tuple, encryption_tuple=encryption_tuple)

    progress_cb_with_timeout = None
    try:
      last_bytes_written = long(0)
      while True:
        with self._ApitoolsRequestHeaders(crypto_headers):
          apitools_request = apitools_messages.StorageObjectsRewriteRequest(
              sourceBucket=src_obj_metadata.bucket,
              sourceObject=src_obj_metadata.name,
              destinationBucket=dst_obj_metadata.bucket,
              # TODO(KMS): Remove the destinationKmsKeyName parameter once the
              # API begins pulling key name from the dest obj metadata.
              destinationKmsKeyName=dst_obj_metadata.kmsKeyName,
              destinationObject=dst_obj_metadata.name,
              projection=projection,
              object=dst_obj_metadata,
              sourceGeneration=src_generation,
              ifGenerationMatch=preconditions.gen_match,
              ifMetagenerationMatch=preconditions.meta_gen_match,
              destinationPredefinedAcl=predefined_acl,
              rewriteToken=resume_rewrite_token,
              maxBytesRewrittenPerCall=max_bytes_per_call,
              userProject=self.user_project)
          rewrite_response = self.api_client.objects.Rewrite(
              apitools_request, global_params=global_params)
        bytes_written = long(rewrite_response.totalBytesRewritten)
        if progress_callback and not progress_cb_with_timeout:
          progress_cb_with_timeout = ProgressCallbackWithTimeout(
              long(rewrite_response.objectSize), progress_callback)
        if progress_cb_with_timeout:
          progress_cb_with_timeout.Progress(bytes_written - last_bytes_written)

        if rewrite_response.done:
          break
        elif not resume_rewrite_token:
          # Save the token and make a tracker file if they don't already exist.
          resume_rewrite_token = rewrite_response.rewriteToken
          WriteRewriteTrackerFile(tracker_file_name, rewrite_params_hash,
                                  rewrite_response.rewriteToken)
        last_bytes_written = bytes_written

      DeleteTrackerFile(tracker_file_name)
      return rewrite_response.resource
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      not_found_exception = CreateNotFoundExceptionForObjectWrite(
          self.provider,
          dst_obj_metadata.bucket,
          src_provider=self.provider,
          src_bucket_name=src_obj_metadata.bucket,
          src_object_name=src_obj_metadata.name,
          src_generation=src_generation)
      self._TranslateExceptionAndRaise(e,
                                       bucket_name=dst_obj_metadata.bucket,
                                       object_name=dst_obj_metadata.name,
                                       not_found_exception=not_found_exception)

  def DeleteObject(self,
                   bucket_name,
                   object_name,
                   preconditions=None,
                   generation=None,
                   provider=None):
    """See CloudApi class for function doc strings."""
    if not preconditions:
      preconditions = Preconditions()

    if generation:
      generation = long(generation)

    apitools_request = apitools_messages.StorageObjectsDeleteRequest(
        bucket=bucket_name,
        object=object_name,
        generation=generation,
        ifGenerationMatch=preconditions.gen_match,
        ifMetagenerationMatch=preconditions.meta_gen_match,
        userProject=self.user_project)
    try:
      return self.api_client.objects.Delete(apitools_request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e,
                                       bucket_name=bucket_name,
                                       object_name=object_name,
                                       generation=generation)

  def ComposeObject(self,
                    src_objs_metadata,
                    dst_obj_metadata,
                    preconditions=None,
                    encryption_tuple=None,
                    provider=None,
                    fields=None):
    """See CloudApi class for function doc strings."""
    ValidateDstObjectMetadata(dst_obj_metadata)

    dst_obj_name = dst_obj_metadata.name
    dst_obj_metadata.name = None
    dst_bucket_name = dst_obj_metadata.bucket
    dst_obj_metadata.bucket = None
    if not dst_obj_metadata.contentType:
      dst_obj_metadata.contentType = DEFAULT_CONTENT_TYPE

    if not preconditions:
      preconditions = Preconditions()

    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))

    src_objs_compose_request = apitools_messages.ComposeRequest(
        sourceObjects=src_objs_metadata, destination=dst_obj_metadata)

    encryption_headers = self._EncryptionHeadersFromTuple(
        crypto_tuple=encryption_tuple)

    kmsKeyName = None
    if encryption_tuple:
      if encryption_tuple.crypto_type == CryptoKeyType.CMEK:
        kmsKeyName = encryption_tuple.crypto_key

    with self._ApitoolsRequestHeaders(encryption_headers):
      apitools_request = apitools_messages.StorageObjectsComposeRequest(
          composeRequest=src_objs_compose_request,
          destinationBucket=dst_bucket_name,
          destinationObject=dst_obj_name,
          ifGenerationMatch=preconditions.gen_match,
          ifMetagenerationMatch=preconditions.meta_gen_match,
          userProject=self.user_project,
          kmsKeyName=kmsKeyName)
      try:
        return self.api_client.objects.Compose(apitools_request,
                                               global_params=global_params)
      except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
        # We can't be sure which object was missing in the 404 case.
        if (isinstance(e, apitools_exceptions.HttpError) and
            e.status_code == 404):
          raise NotFoundException('One of the source objects does not exist.')
        else:
          self._TranslateExceptionAndRaise(e)

  def WatchBucket(self,
                  bucket_name,
                  address,
                  channel_id,
                  token=None,
                  provider=None,
                  fields=None):
    """See CloudApi class for function doc strings."""
    projection = (apitools_messages.StorageObjectsWatchAllRequest.
                  ProjectionValueValuesEnum.full)

    channel = apitools_messages.Channel(address=address,
                                        id=channel_id,
                                        token=token,
                                        type='WEB_HOOK')

    apitools_request = apitools_messages.StorageObjectsWatchAllRequest(
        bucket=bucket_name,
        channel=channel,
        projection=projection,
        userProject=self.user_project)

    global_params = apitools_messages.StandardQueryParameters()
    if fields:
      global_params.fields = ','.join(set(fields))

    try:
      return self.api_client.objects.WatchAll(apitools_request,
                                              global_params=global_params)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def StopChannel(self, channel_id, resource_id, provider=None):
    """See CloudApi class for function doc strings."""
    channel = apitools_messages.Channel(id=channel_id, resourceId=resource_id)
    try:
      self.api_client.channels.Stop(channel)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e)

  def ListChannels(self, bucket_name, provider=None, fields=None):
    """See CloudApi class for function doc strings."""
    apitools_request = apitools_messages.StorageBucketsListChannelsRequest(
        bucket=bucket_name, userProject=self.user_project)

    try:
      return self.api_client.buckets.ListChannels(apitools_request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e, bucket_name=bucket_name)

  def GetProjectServiceAccount(self, project_number):
    """See CloudApi class for function doc strings."""
    try:
      request = apitools_messages.StorageProjectsServiceAccountGetRequest(
          projectId=six.text_type(project_number))
      return self.api_client.projects_serviceAccount.Get(request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e)

  def CreateNotificationConfig(self,
                               bucket_name,
                               pubsub_topic,
                               payload_format,
                               event_types=None,
                               custom_attributes=None,
                               object_name_prefix=None):
    """See CloudApi class for function doc strings."""
    try:
      notification = apitools_messages.Notification(
          topic=pubsub_topic, payload_format=payload_format)
      if event_types:
        notification.event_types = event_types
      if custom_attributes:
        additional_properties = [
            NotificationAdditionalProperty(key=key, value=value)
            for key, value in custom_attributes.items()
        ]
        notification.custom_attributes = NotificationCustomAttributesValue(
            additionalProperties=additional_properties)
      if object_name_prefix:
        notification.object_name_prefix = object_name_prefix

      request = apitools_messages.StorageNotificationsInsertRequest(
          bucket=bucket_name,
          notification=notification,
          userProject=self.user_project)
      return self.api_client.notifications.Insert(request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e)

  def DeleteNotificationConfig(self, bucket_name, notification):
    """See CloudApi class for function doc strings."""
    try:
      request = apitools_messages.StorageNotificationsDeleteRequest(
          bucket=bucket_name,
          notification=notification,
          userProject=self.user_project)
      return self.api_client.notifications.Delete(request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e)

  def ListNotificationConfigs(self, bucket_name):
    """See CloudApi class for function doc strings."""
    try:
      request = apitools_messages.StorageNotificationsListRequest(
          bucket=bucket_name, userProject=self.user_project)
      response = self.api_client.notifications.List(request)
      for notification in response.items:
        yield notification
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e)

  def CreateHmacKey(self, project_id, service_account_email):
    """See CloudApi class for function doc strings."""
    try:
      request = apitools_messages.StorageProjectsHmacKeysCreateRequest(
          projectId=project_id, serviceAccountEmail=service_account_email)
      return self.api_client.hmacKeys.Create(request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e)

  def DeleteHmacKey(self, project_id, access_id):
    """See CloudApi class for function doc strings."""
    try:
      request = apitools_messages.StorageProjectsHmacKeysDeleteRequest(
          projectId=project_id, accessId=access_id)
      return self.api_client.hmacKeys.Delete(request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e)

  def GetHmacKey(self, project_id, access_id):
    """See CloudApi class for function doc strings."""
    try:
      request = apitools_messages.StorageProjectsHmacKeysGetRequest(
          projectId=project_id, accessId=access_id)
      return self.api_client.hmacKeys.Get(request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e)

  def ListHmacKeys(self,
                   project_id,
                   service_account_email,
                   show_deleted_keys=True):
    """See CloudApi class for function doc strings."""
    try:
      request = apitools_messages.StorageProjectsHmacKeysListRequest(
          projectId=project_id,
          serviceAccountEmail=service_account_email,
          showDeletedKeys=show_deleted_keys,
          maxResults=NUM_BUCKETS_PER_LIST_PAGE)

      response = self.api_client.hmacKeys.List(request)
      for key in response.items:
        yield key
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e)

    while response.nextPageToken:
      request = apitools_messages.StorageProjectsHmacKeysListRequest(
          projectId=project_id,
          serviceAccountEmail=service_account_email,
          showDeletedKeys=show_deleted_keys,
          pageToken=response.nextPageToken,
          maxResults=NUM_BUCKETS_PER_LIST_PAGE)
      try:
        response = self.api_client.hmacKeys.List(request)
      except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
        self._TranslateExceptionAndRaise(e)

      for key in response.items:
        yield key

  def UpdateHmacKey(self, project_id, access_id, state, etag):
    """See CloudApi class for function doc strings."""
    try:
      request = apitools_messages.StorageProjectsHmacKeysUpdateRequest(
          projectId=project_id, accessId=access_id)
      metadata = apitools_messages.HmacKeyMetadata()
      metadata.state = state
      if etag:
        metadata.etag = etag
      request.resource = metadata
      return self.api_client.hmacKeys.Update(request)
    except TRANSLATABLE_APITOOLS_EXCEPTIONS as e:
      self._TranslateExceptionAndRaise(e)

  def _BucketCannedAclToPredefinedAcl(self, canned_acl_string):
    """Translates the input string to a bucket PredefinedAcl string.

    Args:
      canned_acl_string: Canned ACL string.

    Returns:
      String that can be used as a query parameter with the JSON API. This
      corresponds to a flavor of *PredefinedAclValueValuesEnum and can be
      used as input to apitools requests that affect bucket access controls.
    """
    # XML : JSON
    translation_dict = {
        None: None,
        'authenticated-read': 'authenticatedRead',
        'private': 'private',
        'project-private': 'projectPrivate',
        'public-read': 'publicRead',
        'public-read-write': 'publicReadWrite'
    }
    if canned_acl_string in translation_dict:
      return translation_dict[canned_acl_string]
    raise ArgumentException('Invalid canned ACL %s' % canned_acl_string)

  def _ObjectCannedAclToPredefinedAcl(self, canned_acl_string):
    """Translates the input string to an object PredefinedAcl string.

    Args:
      canned_acl_string: Canned ACL string.

    Returns:
      String that can be used as a query parameter with the JSON API. This
      corresponds to a flavor of *PredefinedAclValueValuesEnum and can be
      used as input to apitools requests that affect object access controls.
    """
    # XML : JSON
    translation_dict = {
        None: None,
        'authenticated-read': 'authenticatedRead',
        'bucket-owner-read': 'bucketOwnerRead',
        'bucket-owner-full-control': 'bucketOwnerFullControl',
        'private': 'private',
        'project-private': 'projectPrivate',
        'public-read': 'publicRead'
    }
    if canned_acl_string in translation_dict:
      return translation_dict[canned_acl_string]
    raise ArgumentException('Invalid canned ACL %s' % canned_acl_string)

  def _ValidateHttpAccessTokenRefreshError(self, e):
    if (isinstance(e, oauth2client.client.HttpAccessTokenRefreshError) and
        not (e.status == 429 or e.status >= 500)):
      raise

  def _TranslateExceptionAndRaise(self,
                                  e,
                                  bucket_name=None,
                                  object_name=None,
                                  generation=None,
                                  not_found_exception=None):
    """Translates an HTTP exception and raises the translated or original value.

    Args:
      e: Any Exception.
      bucket_name: Optional bucket name in request that caused the exception.
      object_name: Optional object name in request that caused the exception.
      generation: Optional generation in request that caused the exception.
      not_found_exception: Optional exception to raise in the not-found case.

    Raises:
      Translated CloudApi exception, or the original exception if it was not
      translatable.
    """
    if self.logger.isEnabledFor(logging.DEBUG):
      self.logger.debug('TranslateExceptionAndRaise: %s',
                        traceback.format_exc())
    translated_exception = self._TranslateApitoolsException(
        e,
        bucket_name=bucket_name,
        object_name=object_name,
        generation=generation,
        not_found_exception=not_found_exception)
    if translated_exception:
      raise translated_exception
    else:
      raise

  def _GetMessageFromHttpError(self, http_error):
    if isinstance(http_error, apitools_exceptions.HttpError):
      if getattr(http_error, 'content', None):
        try:
          json_obj = json.loads(six.ensure_str(http_error.content))
          if 'error' in json_obj and 'message' in json_obj['error']:
            return json_obj['error']['message']
        except Exception:  # pylint: disable=broad-except
          # If we couldn't decode anything, just leave the message as None.
          pass

  def _GetAcceptableScopesFromHttpError(self, http_error):
    try:
      www_authenticate = http_error.response['www-authenticate']
      # In the event of a scope error, the www-authenticate field of the HTTP
      # response should contain text of the form
      #
      # 'Bearer realm="https://oauth2.googleapis.com/",
      # error=insufficient_scope, scope="${space separated list of acceptable
      # scopes}"'
      #
      # Here we use a quick string search to find the scope list, just looking
      # for a substring with the form 'scope="${scopes}"'.
      scope_idx = www_authenticate.find('scope="')
      if scope_idx >= 0:
        scopes = www_authenticate[scope_idx:].split('"')[1]
        return 'Acceptable scopes: %s' % scopes
    except Exception:  # pylint: disable=broad-except
      # Return None if we have any trouble parsing out the acceptable scopes.
      pass

  def _TranslateApitoolsResumableUploadException(self, e):
    if isinstance(e, apitools_exceptions.HttpError):
      message = self._GetMessageFromHttpError(e)
      if e.status_code >= 500:
        return ResumableUploadException(message or 'Server Error',
                                        status=e.status_code)
      elif e.status_code == 429:
        return ResumableUploadException(message or 'Too Many Requests',
                                        status=e.status_code)
      elif e.status_code == 410:
        return ResumableUploadStartOverException(message or 'Bad Request',
                                                 status=e.status_code)
      elif e.status_code == 404:
        return ResumableUploadStartOverException(message or 'Bad Request',
                                                 status=e.status_code)
      elif e.status_code >= 400:
        return ResumableUploadAbortException(message or 'Bad Request',
                                             status=e.status_code)
    if isinstance(e, apitools_exceptions.StreamExhausted):
      return ResumableUploadAbortException(
          '%s; if this issue persists, try deleting the tracker files present'
          ' under ~/.gsutil/tracker-files/' % str(e))
    if isinstance(e, apitools_exceptions.TransferError):
      if ('Aborting transfer' in str(e) or
          'Not enough bytes in stream' in str(e)):
        return ResumableUploadAbortException(str(e))
      elif 'additional bytes left in stream' in str(e):
        return ResumableUploadAbortException(
            '%s; this can happen if a file changes size while being uploaded' %
            str(e))

  def _TranslateApitoolsException(self,
                                  e,
                                  bucket_name=None,
                                  object_name=None,
                                  generation=None,
                                  not_found_exception=None):
    """Translates apitools exceptions into their gsutil Cloud Api equivalents.

    Args:
      e: Any exception in TRANSLATABLE_APITOOLS_EXCEPTIONS.
      bucket_name: Optional bucket name in request that caused the exception.
      object_name: Optional object name in request that caused the exception.
      generation: Optional generation in request that caused the exception.
      not_found_exception: Optional exception to raise in the not-found case.

    Returns:
      ServiceException for translatable exceptions, None
      otherwise.
    """
    if isinstance(e, apitools_exceptions.HttpError):
      message = self._GetMessageFromHttpError(e)
      if e.status_code == 400:
        # It is possible that the Project ID is incorrect.  Unfortunately the
        # JSON API does not give us much information about what part of the
        # request was bad.
        return BadRequestException(message or 'Bad Request',
                                   status=e.status_code)
      elif e.status_code == 401:
        if 'Login Required' in str(e):
          return AccessDeniedException(message or
                                       'Access denied: login required.',
                                       status=e.status_code)
        elif 'insufficient_scope' in str(e):
          # If the service includes insufficient scope error detail in the
          # response body, this check can be removed.
          return AccessDeniedException(
              _INSUFFICIENT_OAUTH2_SCOPE_MESSAGE,
              status=e.status_code,
              body=self._GetAcceptableScopesFromHttpError(e))
      elif e.status_code == 403:
        if 'The account for the specified project has been disabled' in str(e):
          return AccessDeniedException(message or 'Account disabled.',
                                       status=e.status_code)
        elif 'Daily Limit for Unauthenticated Use Exceeded' in str(e):
          return AccessDeniedException(message or
                                       'Access denied: quota exceeded. '
                                       'Is your project ID valid?',
                                       status=e.status_code)
        elif 'The bucket you tried to delete was not empty.' in str(e):
          return NotEmptyException('BucketNotEmpty (%s)' % bucket_name,
                                   status=e.status_code)
        elif ('The bucket you tried to create requires domain ownership '
              'verification.' in str(e)):
          return AccessDeniedException(
              'The bucket you tried to create requires domain ownership '
              'verification. Please see '
              'https://cloud.google.com/storage/docs/naming'
              '?hl=en#verification for more details.',
              status=e.status_code)
        elif 'User Rate Limit Exceeded' in str(e):
          return AccessDeniedException(
              'Rate limit exceeded. Please retry this '
              'request later.',
              status=e.status_code)
        elif 'Access Not Configured' in str(e):
          return AccessDeniedException(
              'Access Not Configured. Please go to the Google Cloud Platform '
              'Console (https://cloud.google.com/console#/project) for your '
              'project, select APIs and Auth and enable the '
              'Google Cloud Storage JSON API.',
              status=e.status_code)
        elif 'insufficient_scope' in str(e):
          # If the service includes insufficient scope error detail in the
          # response body, this check can be removed.
          return AccessDeniedException(
              _INSUFFICIENT_OAUTH2_SCOPE_MESSAGE,
              status=e.status_code,
              body=self._GetAcceptableScopesFromHttpError(e))
        elif 'does not have permission to publish messages' in str(e):
          return PublishPermissionDeniedException(message, status=e.status_code)
        else:
          return AccessDeniedException(message or str(e), status=e.status_code)
      elif e.status_code == 404:
        if not_found_exception:
          # The exception is pre-constructed prior to translation; the HTTP
          # status code isn't available at that time.
          setattr(not_found_exception, 'status', e.status_code)
          return not_found_exception
        elif bucket_name:
          if object_name:
            return CreateObjectNotFoundException(e.status_code,
                                                 self.provider,
                                                 bucket_name,
                                                 object_name,
                                                 generation=generation)
          return CreateBucketNotFoundException(e.status_code, self.provider,
                                               bucket_name)
        return NotFoundException(message or e.message, status=e.status_code)

      elif e.status_code == 409 and bucket_name:
        if 'The bucket you tried to delete is not empty.' in str(e):
          return NotEmptyException('BucketNotEmpty (%s)' % bucket_name,
                                   status=e.status_code)
        return ServiceException(
            'A Cloud Storage bucket named \'%s\' already exists. Try another '
            'name. Bucket names must be globally unique across all Google Cloud '
            'projects, including those outside of your '
            'organization.' % bucket_name,
            status=e.status_code)
      elif e.status_code == 412:
        return PreconditionException(message, status=e.status_code)
      return ServiceException(message, status=e.status_code)
    elif isinstance(e, apitools_exceptions.TransferInvalidError):
      return ServiceException('Transfer invalid (possible encoding error: %s)' %
                              str(e)) def StorageUrlFromString(url_str):
  """Static factory function for creating a StorageUrl from a string."""

  scheme = GetSchemeFromUrlString(url_str)

  if not IsKnownUrlScheme(scheme):
    raise InvalidUrlError('Unrecognized scheme "%s"' % scheme)
  if scheme == 'file':
    path = _GetPathFromUrlString(url_str)
    is_stream = (path == '-')
    is_fifo = False
    try:
      is_fifo = stat.S_ISFIFO(os.stat(path).st_mode)
    except OSError:
      pass
    return _FileUrl(url_str, is_stream=is_stream, is_fifo=is_fifo)
  return _CloudUrl(url_str) def SkipForGS(reason):
  if not RUN_S3_TESTS:
    return unittest.skip(reason)
  else:
    return lambda func: func def SkipForS3(reason):
  if RUN_S3_TESTS:
    return unittest.skip(reason)
  else:
    return lambda func: func def SkipForXML(reason):
  """Skips the test if running S3 tests, or if prefer_api isn't set to json."""
  if not USING_JSON_API or RUN_S3_TESTS:
    return unittest.skip(reason)
  else:
    return lambda func: func def SkipForJSON(reason):
  if USING_JSON_API:
    return unittest.skip(reason)
  else:
    return lambda func: func class Bucket(_messages.Message):
  class AutoclassValue(_messages.Message):
  class BillingValue(_messages.Message):
  class CorsValueListEntry(_messages.Message):
  class CustomPlacementConfigValue(_messages.Message):
  class EncryptionValue(_messages.Message):
  class IamConfigurationValue(_messages.Message):
    class BucketPolicyOnlyValue(_messages.Message):
  class LabelsValue(_messages.Message):
    class AdditionalProperty(_messages.Message):
  class LifecycleValue(_messages.Message):
    class RuleValueListEntry(_messages.Message):
      class ActionValue(_messages.Message):
      class ConditionValue(_messages.Message):
  class LoggingValue(_messages.Message):
  class OwnerValue(_messages.Message):
  class RetentionPolicyValue(_messages.Message):
  class VersioningValue(_messages.Message):
  class WebsiteValue(_messages.Message):
class BucketAccessControl(_messages.Message):
  class ProjectTeamValue(_messages.Message):
class BucketAccessControls(_messages.Message):
class Buckets(_messages.Message):
class Channel(_messages.Message):
  class ParamsValue(_messages.Message):
    class AdditionalProperty(_messages.Message):
class Channels(_messages.Message):
  class ItemsValueListEntry(_messages.Message):
class ComposeRequest(_messages.Message):
  class SourceObjectsValueListEntry(_messages.Message):
    class ObjectPreconditionsValue(_messages.Message):
class Expr(_messages.Message):
class Notification(_messages.Message):
  class CustomAttributesValue(_messages.Message):
    class AdditionalProperty(_messages.Message):
class Notifications(_messages.Message):
class Object(_messages.Message):
  class CustomerEncryptionValue(_messages.Message):
  class MetadataValue(_messages.Message):
    class AdditionalProperty(_messages.Message):
  class OwnerValue(_messages.Message):
class ObjectAccessControl(_messages.Message):
  class ProjectTeamValue(_messages.Message):
class ObjectAccessControls(_messages.Message):
class Objects(_messages.Message):
class Policy(_messages.Message):
  class BindingsValueListEntry(_messages.Message):
class RewriteResponse(_messages.Message):
class ServiceAccount(_messages.Message):
class StandardQueryParameters(_messages.Message):
  class AltValueValuesEnum(_messages.Enum):
class StorageBucketAccessControlsDeleteRequest(_messages.Message):
class StorageBucketAccessControlsDeleteResponse(_messages.Message):
class StorageBucketAccessControlsGetRequest(_messages.Message):
class StorageBucketAccessControlsInsertRequest(_messages.Message):
class StorageBucketAccessControlsListRequest(_messages.Message):
class StorageBucketAccessControlsPatchRequest(_messages.Message):
class StorageBucketAccessControlsUpdateRequest(_messages.Message):
class StorageBucketsDeleteRequest(_messages.Message):
class StorageBucketsDeleteResponse(_messages.Message):
class StorageBucketsGetIamPolicyRequest(_messages.Message):
class StorageBucketsGetRequest(_messages.Message):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageBucketsInsertRequest(_messages.Message):
  class PredefinedAclValueValuesEnum(_messages.Enum):
  class PredefinedDefaultObjectAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageBucketsListRequest(_messages.Message):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageBucketsListChannelsRequest(_messages.Message):
class StorageBucketsLockRetentionPolicyRequest(_messages.Message):
class StorageBucketsPatchRequest(_messages.Message):
  class PredefinedAclValueValuesEnum(_messages.Enum):
  class PredefinedDefaultObjectAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageBucketsSetIamPolicyRequest(_messages.Message):
class StorageBucketsTestIamPermissionsRequest(_messages.Message):
class StorageBucketsUpdateRequest(_messages.Message):
  class PredefinedAclValueValuesEnum(_messages.Enum):
  class PredefinedDefaultObjectAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageChannelsStopResponse(_messages.Message):
class StorageDefaultObjectAccessControlsDeleteRequest(_messages.Message):
class StorageDefaultObjectAccessControlsDeleteResponse(_messages.Message):
class StorageDefaultObjectAccessControlsGetRequest(_messages.Message):
class StorageDefaultObjectAccessControlsInsertRequest(_messages.Message):
class StorageDefaultObjectAccessControlsListRequest(_messages.Message):
class StorageDefaultObjectAccessControlsPatchRequest(_messages.Message):
class StorageDefaultObjectAccessControlsUpdateRequest(_messages.Message):
class StorageNotificationsDeleteRequest(_messages.Message):
class StorageNotificationsDeleteResponse(_messages.Message):
class StorageNotificationsGetRequest(_messages.Message):
class StorageNotificationsInsertRequest(_messages.Message):
class StorageNotificationsListRequest(_messages.Message):
class StorageObjectAccessControlsDeleteRequest(_messages.Message):
class StorageObjectAccessControlsDeleteResponse(_messages.Message):
class StorageObjectAccessControlsGetRequest(_messages.Message):
class StorageObjectAccessControlsInsertRequest(_messages.Message):
class StorageObjectAccessControlsListRequest(_messages.Message):
class StorageObjectAccessControlsPatchRequest(_messages.Message):
class StorageObjectAccessControlsUpdateRequest(_messages.Message):
class StorageObjectsComposeRequest(_messages.Message):
  class DestinationPredefinedAclValueValuesEnum(_messages.Enum):
class StorageObjectsCopyRequest(_messages.Message):
  class DestinationPredefinedAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsDeleteRequest(_messages.Message):
class StorageObjectsDeleteResponse(_messages.Message):
class StorageObjectsGetIamPolicyRequest(_messages.Message):
class StorageObjectsGetRequest(_messages.Message):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsInsertRequest(_messages.Message):
  class PredefinedAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsListRequest(_messages.Message):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsPatchRequest(_messages.Message):
  class PredefinedAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsRewriteRequest(_messages.Message):
  class DestinationPredefinedAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsSetIamPolicyRequest(_messages.Message):
class StorageObjectsTestIamPermissionsRequest(_messages.Message):
class StorageObjectsUpdateRequest(_messages.Message):
  class PredefinedAclValueValuesEnum(_messages.Enum):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageObjectsWatchAllRequest(_messages.Message):
  class ProjectionValueValuesEnum(_messages.Enum):
class StorageProjectsServiceAccountGetRequest(_messages.Message):
class TestIamPermissionsResponse(_messages.Message):
class StorageProjectsHmacKeysCreateRequest(_messages.Message):
class StorageProjectsHmacKeysDeleteRequest(_messages.Message):
class StorageProjectsHmacKeysGetRequest(_messages.Message):
class StorageProjectsHmacKeysListRequest(_messages.Message):
class StorageProjectsHmacKeysUpdateRequest(_messages.Message):
class HmacKeyMetadata(_messages.Message):
class HmacKey(_messages.Message):
class HmacKeyMetadataList(_messages.Message):
class HmacKeysDeleteResponse(_messages.Message): def UsingCrcmodExtension():
  boto_opt = boto.config.get('GSUtil', 'test_assume_fast_crcmod', None)
  if boto_opt is not None:
    return boto_opt
  # Python 3 makes this attribute tough to access due to the way the top-level
  # crcmod package imports and (identically) names its crcmod module. The only
  # way to get it is "from crcmod.crcmod import _usingExtension". This is the
  # alternative form of that statement, but doesn't pollute this module's
  # namespace with a "_usingExtension" attribute. This also works in both Python
  # 2.7 and 3.5+.
  nested_crcmod = __import__(
      'crcmod.crcmod',
      globals(),
      locals(),
      ['_usingExtension'],
      0,
  )
  return getattr(nested_crcmod, '_usingExtension', False) _RENAME_ON_HASH_MISMATCH = False
_RENAME_ON_HASH_MISMATCH_SUFFIX = '_corrupt'
PARALLEL_UPLOAD_TEMP_NAMESPACE = (
    '/gsutil/tmp/parallel_composite_uploads/for_details_see/gsutil_help_cp/')
PARALLEL_UPLOAD_STATIC_SALT = u"""
PARALLEL_UPLOAD_SALT_TO_PREVENT_COLLISIONS.
The theory is that no user will have prepended this to the front of
one of their object names and then done an MD5 hash of the name, and
then prepended PARALLEL_UPLOAD_TEMP_NAMESPACE to the front of their object
name. Note that there will be no problems with object name length since we
hash the original name.
"""
UPLOAD_RETURN_FIELDS = [
    'crc32c',
    'customerEncryption',
    'etag',
    'generation',
    'md5Hash',
    'size',
]
GZIP_CHUNK_SIZE = 8192
GZIP_ALL_FILES = 'GZIP_ALL_FILES'
TRACKERFILE_UPDATE_THRESHOLD = TEN_MIB
PARALLEL_COMPOSITE_SUGGESTION_THRESHOLD = 150 * 1024 * 1024
S3_MAX_UPLOAD_SIZE = 5 * 1024 * 1024 * 1024
COMMON_EXTENSION_RULES = {
    'md': 'text/markdown',
    'tgz': 'application/gzip',
}
EARLY_DELETION_MINIMUM_LIFETIME = {
    'nearline': 30 * SECONDS_PER_DAY,
    'coldline': 90 * SECONDS_PER_DAY,
    'archive': 365 * SECONDS_PER_DAY
}
class FileConcurrencySkipError(Exception):
class SlicedDownloadFileWrapper(object):
class Manifest(object):
class ItemExistsError(Exception):
class SkipUnsupportedObjectError(Exception):
class SkipGlacierError(SkipUnsupportedObjectError):
def _RmExceptionHandler(cls, e):
def _ParallelCopyExceptionHandler(cls, e):
def _PerformParallelUploadFileToObject(cls, args, thread_state=None):
def CreateCopyHelperOpts(perform_mv=False,
                         no_clobber=False,
                         daisy_chain=False,
                         read_args_from_stdin=False,
                         print_ver=False,
                         use_manifest=False,
                         preserve_acl=False,
                         canned_acl=None,
                         skip_unsupported_objects=False,
                         test_callback_file=None,
                         dest_storage_class=None):
def GetCopyHelperOpts():
def _SelectDownloadStrategy(dst_url):
def InsistDstUrlNamesContainer(exp_dst_url, have_existing_dst_container,
                               command_name):
def _ShouldTreatDstUrlAsBucketSubDir(have_multiple_srcs, dst_url,
                                     have_existing_dest_subdir,
                                     src_url_names_container,
                                     recursion_requested):
def _ShouldTreatDstUrlAsSingleton(src_url_names_container, have_multiple_srcs,
                                  have_existing_dest_subdir, dst_url,
                                  recursion_requested):
def ConstructDstUrl(src_url,
                    exp_src_url,
                    src_url_names_container,
                    have_multiple_srcs,
                    exp_dst_url,
                    have_existing_dest_subdir,
                    recursion_requested,
                    preserve_posix=False):
def _CreateDigestsFromDigesters(digesters):
def _CreateDigestsFromLocalFile(status_queue, algs, file_name, src_url,
                                src_obj_metadata):
def _CheckCloudHashes(logger, src_url, dst_url, src_obj_metadata,
                      dst_obj_metadata):
def _CheckHashes(logger,
                 obj_url,
                 obj_metadata,
                 file_name,
                 digests,
                 is_upload=False):
def IsNoClobberServerException(e):
def CheckForDirFileConflict(exp_src_url, dst_url):
def _PartitionFile(fp,
                   file_size,
                   src_url,
                   content_type,
                   canned_acl,
                   dst_bucket_url,
                   random_prefix,
                   tracker_file,
                   tracker_file_lock,
                   encryption_key_sha256=None,
                   gzip_encoded=False):
def _GetComponentNumber(component):
def _DoParallelCompositeUpload(fp,
                               src_url,
                               dst_url,
                               dst_obj_metadata,
                               canned_acl,
                               file_size,
                               preconditions,
                               gsutil_api,
                               command_obj,
                               copy_exception_handler,
                               logger,
                               gzip_encoded=False):
def _ShouldDoParallelCompositeUpload(logger,
                                     allow_splitting,
                                     src_url,
                                     dst_url,
                                     file_size,
                                     gsutil_api,
                                     canned_acl=None):
def ExpandUrlToSingleBlr(url_str,
                         gsutil_api,
                         project_id,
                         treat_nonexistent_object_as_subdir=False,
                         logger=None):
def FixWindowsNaming(src_url, dst_url):
def SrcDstSame(src_url, dst_url):
def _LogCopyOperation(logger, src_url, dst_url, dst_obj_metadata):
def _CopyObjToObjInTheCloud(src_url,
                            src_obj_metadata,
                            dst_url,
                            dst_obj_metadata,
                            preconditions,
                            gsutil_api,
                            decryption_key=None):
def _SetContentTypeFromFile(src_url, dst_obj_metadata):
def _UploadFileToObjectNonResumable(src_url,
                                    src_obj_filestream,
                                    src_obj_size,
                                    dst_url,
                                    dst_obj_metadata,
                                    preconditions,
                                    gsutil_api,
                                    gzip_encoded=False):
def _UploadFileToObjectResumable(src_url,
                                 src_obj_filestream,
                                 src_obj_size,
                                 dst_url,
                                 dst_obj_metadata,
                                 preconditions,
                                 gsutil_api,
                                 logger,
                                 is_component=False,
                                 gzip_encoded=False):
  def _UploadTrackerCallback(serialization_data):
def _SelectUploadCompressionStrategy(object_name,
                                     is_component=False,
                                     gzip_exts=False,
                                     gzip_encoded=False):
def _ApplyZippedUploadCompression(src_url, src_obj_filestream, src_obj_size,
                                  logger):
def _DelegateUploadFileToObject(upload_delegate, upload_url, upload_stream,
                                zipped_file, gzip_encoded_file,
                                parallel_composite_upload, logger):
def _UploadFileToObject(src_url,
                        src_obj_filestream,
                        src_obj_size,
                        dst_url,
                        dst_obj_metadata,
                        preconditions,
                        gsutil_api,
                        logger,
                        command_obj,
                        copy_exception_handler,
                        gzip_exts=None,
                        allow_splitting=True,
                        is_component=False,
                        gzip_encoded=False):
  def CallParallelCompositeUpload():
  def CallNonResumableUpload():
  def CallResumableUpload():
def _GetDownloadFile(dst_url, src_obj_metadata, logger):
def _ShouldDoSlicedDownload(download_strategy, src_obj_metadata,
                            allow_splitting, logger):
def _PerformSlicedDownloadObjectToFile(cls, args, thread_state=None):
def _MaintainSlicedDownloadTrackerFiles(src_obj_metadata, dst_url,
                                        download_file_name, logger,
                                        api_selector, num_components):
  def __init__(self, fp, tracker_file_name, src_obj_metadata, start_byte,
               end_byte):
  def mode(self):
  def write(self, data):  # pylint: disable=invalid-name
  def seek(self, offset, whence=os.SEEK_SET):  # pylint: disable=invalid-name
  def tell(self):  # pylint: disable=invalid-name
  def flush(self):  # pylint: disable=invalid-name
  def close(self):  # pylint: disable=invalid-name
def _PartitionObject(src_url,
                     src_obj_metadata,
                     dst_url,
                     download_file_name,
                     decryption_key=None):
def _DoSlicedDownload(src_url,
                      src_obj_metadata,
                      dst_url,
                      download_file_name,
                      command_obj,
                      logger,
                      copy_exception_handler,
                      api_selector,
                      decryption_key=None,
                      status_queue=None):
def _DownloadObjectToFileResumable(src_url,
                                   src_obj_metadata,
                                   dst_url,
                                   download_file_name,
                                   gsutil_api,
                                   logger,
                                   digesters,
                                   component_num=None,
                                   start_byte=0,
                                   end_byte=None,
                                   decryption_key=None):
def _DownloadObjectToFileNonResumable(src_url,
                                      src_obj_metadata,
                                      dst_url,
                                      download_file_name,
                                      gsutil_api,
                                      digesters,
                                      decryption_key=None):
def _DownloadObjectToFile(src_url,
                          src_obj_metadata,
                          dst_url,
                          gsutil_api,
                          logger,
                          command_obj,
                          copy_exception_handler,
                          allow_splitting=True,
                          decryption_key=None,
                          is_rsync=False,
                          preserve_posix=False,
                          use_stet=False):
def _ValidateAndCompleteDownload(logger,
                                 src_url,
                                 src_obj_metadata,
                                 dst_url,
                                 need_to_unzip,
                                 server_gzip,
                                 digesters,
                                 hash_algs,
                                 temporary_file_name,
                                 api_selector,
                                 bytes_transferred,
                                 gsutil_api,
                                 is_rsync=False,
                                 preserve_posix=False,
                                 use_stet=False):
def _CopyFileToFile(src_url, dst_url, status_queue=None, src_obj_metadata=None):
def _DummyTrackerCallback(_):
def _CopyObjToObjDaisyChainMode(src_url,
                                src_obj_metadata,
                                dst_url,
                                dst_obj_metadata,
                                preconditions,
                                gsutil_api,
                                logger,
                                decryption_key=None):
def GetSourceFieldsNeededForCopy(dst_is_cloud,
                                 skip_unsupported_objects,
                                 preserve_acl,
                                 is_rsync=False,
                                 preserve_posix=False,
                                 delete_source=False,
                                 file_size_will_change=False):
def WarnIfMvEarlyDeletionChargeApplies(src_url, src_obj_metadata, logger):
def MaybeSkipUnsupportedObject(src_url, src_obj_metadata):
def GetDecryptionCSEK(src_url, src_obj_metadata):
def PerformCopy(logger,
                src_url,
                dst_url,
                gsutil_api,
                command_obj,
                copy_exception_handler,
                src_obj_metadata=None,
                allow_splitting=True,
                headers=None,
                manifest=None,
                gzip_exts=None,
                is_rsync=False,
                preserve_posix=False,
                gzip_encoded=False,
                use_stet=False):
  def __init__(self, path):
  def _ParseManifest(self):
  def WasSuccessful(self, src):
  def _CreateManifestFile(self):
  def Set(self, url, key, value):
  def Initialize(self, source_url, destination_url):
  def SetResult(self, source_url, bytes_transferred, result, description=''):
  def _WriteRowToManifestFile(self, url):
  def _RemoveItemFromManifest(self, url):
  def __init__(self):
  def __init__(self):
def GetPathBeforeFinalDir(url, exp_src_url):
def ResolveWildcardsInPathBeforeFinalDir(src_url_path_sans_final_dir,
                                         exp_src_url):
def _GetPartitionInfo(file_size, max_components, default_component_size):
def _DeleteTempComponentObjectFn(cls, url_to_delete, thread_state=None):
def FilterExistingComponents(dst_args, existing_components, bucket_url,
                             gsutil_api): PARALLEL_UPLOAD_STATIC_SALT = u"""
PARALLEL_UPLOAD_SALT_TO_PREVENT_COLLISIONS.
The theory is that no user will have prepended this to the front of
one of their object names and then done an MD5 hash of the name, and
then prepended PARALLEL_UPLOAD_TEMP_NAMESPACE to the front of their object
name. Note that there will be no problems with object name length since we
hash the original name.
""" PARALLEL_UPLOAD_TEMP_NAMESPACE = (
    '/gsutil/tmp/parallel_composite_uploads/for_details_see/gsutil_help_cp/') _RENAME_ON_HASH_MISMATCH = False
_RENAME_ON_HASH_MISMATCH_SUFFIX = '_corrupt'
PARALLEL_UPLOAD_TEMP_NAMESPACE = (
    '/gsutil/tmp/parallel_composite_uploads/for_details_see/gsutil_help_cp/')
PARALLEL_UPLOAD_STATIC_SALT = u"""
PARALLEL_UPLOAD_SALT_TO_PREVENT_COLLISIONS.
The theory is that no user will have prepended this to the front of
one of their object names and then done an MD5 hash of the name, and
then prepended PARALLEL_UPLOAD_TEMP_NAMESPACE to the front of their object
name. Note that there will be no problems with object name length since we
hash the original name.
"""
UPLOAD_RETURN_FIELDS = [
    'crc32c',
    'customerEncryption',
    'etag',
    'generation',
    'md5Hash',
    'size',
]
GZIP_CHUNK_SIZE = 8192
GZIP_ALL_FILES = 'GZIP_ALL_FILES'
TRACKERFILE_UPDATE_THRESHOLD = TEN_MIB
PARALLEL_COMPOSITE_SUGGESTION_THRESHOLD = 150 * 1024 * 1024
S3_MAX_UPLOAD_SIZE = 5 * 1024 * 1024 * 1024
COMMON_EXTENSION_RULES = {
    'md': 'text/markdown',
    'tgz': 'application/gzip',
}
EARLY_DELETION_MINIMUM_LIFETIME = {
    'nearline': 30 * SECONDS_PER_DAY,
    'coldline': 90 * SECONDS_PER_DAY,
    'archive': 365 * SECONDS_PER_DAY
}
class FileConcurrencySkipError(Exception):
class SlicedDownloadFileWrapper(object):
class Manifest(object):
class ItemExistsError(Exception):
class SkipUnsupportedObjectError(Exception):
class SkipGlacierError(SkipUnsupportedObjectError):
def _RmExceptionHandler(cls, e):
def _ParallelCopyExceptionHandler(cls, e):
def _PerformParallelUploadFileToObject(cls, args, thread_state=None):
def CreateCopyHelperOpts(perform_mv=False,
                         no_clobber=False,
                         daisy_chain=False,
                         read_args_from_stdin=False,
                         print_ver=False,
                         use_manifest=False,
                         preserve_acl=False,
                         canned_acl=None,
                         skip_unsupported_objects=False,
                         test_callback_file=None,
                         dest_storage_class=None):
def GetCopyHelperOpts():
def _SelectDownloadStrategy(dst_url):
def InsistDstUrlNamesContainer(exp_dst_url, have_existing_dst_container,
                               command_name):
def _ShouldTreatDstUrlAsBucketSubDir(have_multiple_srcs, dst_url,
                                     have_existing_dest_subdir,
                                     src_url_names_container,
                                     recursion_requested):
def _ShouldTreatDstUrlAsSingleton(src_url_names_container, have_multiple_srcs,
                                  have_existing_dest_subdir, dst_url,
                                  recursion_requested):
def ConstructDstUrl(src_url,
                    exp_src_url,
                    src_url_names_container,
                    have_multiple_srcs,
                    exp_dst_url,
                    have_existing_dest_subdir,
                    recursion_requested,
                    preserve_posix=False):
def _CreateDigestsFromDigesters(digesters):
def _CreateDigestsFromLocalFile(status_queue, algs, file_name, src_url,
                                src_obj_metadata):
def _CheckCloudHashes(logger, src_url, dst_url, src_obj_metadata,
                      dst_obj_metadata):
def _CheckHashes(logger,
                 obj_url,
                 obj_metadata,
                 file_name,
                 digests,
                 is_upload=False):
def IsNoClobberServerException(e):
def CheckForDirFileConflict(exp_src_url, dst_url):
def _PartitionFile(fp,
                   file_size,
                   src_url,
                   content_type,
                   canned_acl,
                   dst_bucket_url,
                   random_prefix,
                   tracker_file,
                   tracker_file_lock,
                   encryption_key_sha256=None,
                   gzip_encoded=False):
def _GetComponentNumber(component):
def _DoParallelCompositeUpload(fp,
                               src_url,
                               dst_url,
                               dst_obj_metadata,
                               canned_acl,
                               file_size,
                               preconditions,
                               gsutil_api,
                               command_obj,
                               copy_exception_handler,
                               logger,
                               gzip_encoded=False):
def _ShouldDoParallelCompositeUpload(logger,
                                     allow_splitting,
                                     src_url,
                                     dst_url,
                                     file_size,
                                     gsutil_api,
                                     canned_acl=None):
def ExpandUrlToSingleBlr(url_str,
                         gsutil_api,
                         project_id,
                         treat_nonexistent_object_as_subdir=False,
                         logger=None):
def FixWindowsNaming(src_url, dst_url):
def SrcDstSame(src_url, dst_url):
def _LogCopyOperation(logger, src_url, dst_url, dst_obj_metadata):
def _CopyObjToObjInTheCloud(src_url,
                            src_obj_metadata,
                            dst_url,
                            dst_obj_metadata,
                            preconditions,
                            gsutil_api,
                            decryption_key=None):
def _SetContentTypeFromFile(src_url, dst_obj_metadata):
def _UploadFileToObjectNonResumable(src_url,
                                    src_obj_filestream,
                                    src_obj_size,
                                    dst_url,
                                    dst_obj_metadata,
                                    preconditions,
                                    gsutil_api,
                                    gzip_encoded=False):
def _UploadFileToObjectResumable(src_url,
                                 src_obj_filestream,
                                 src_obj_size,
                                 dst_url,
                                 dst_obj_metadata,
                                 preconditions,
                                 gsutil_api,
                                 logger,
                                 is_component=False,
                                 gzip_encoded=False):
  def _UploadTrackerCallback(serialization_data):
def _SelectUploadCompressionStrategy(object_name,
                                     is_component=False,
                                     gzip_exts=False,
                                     gzip_encoded=False):
def _ApplyZippedUploadCompression(src_url, src_obj_filestream, src_obj_size,
                                  logger):
def _DelegateUploadFileToObject(upload_delegate, upload_url, upload_stream,
                                zipped_file, gzip_encoded_file,
                                parallel_composite_upload, logger):
def _UploadFileToObject(src_url,
                        src_obj_filestream,
                        src_obj_size,
                        dst_url,
                        dst_obj_metadata,
                        preconditions,
                        gsutil_api,
                        logger,
                        command_obj,
                        copy_exception_handler,
                        gzip_exts=None,
                        allow_splitting=True,
                        is_component=False,
                        gzip_encoded=False):
  def CallParallelCompositeUpload():
  def CallNonResumableUpload():
  def CallResumableUpload():
def _GetDownloadFile(dst_url, src_obj_metadata, logger):
def _ShouldDoSlicedDownload(download_strategy, src_obj_metadata,
                            allow_splitting, logger):
def _PerformSlicedDownloadObjectToFile(cls, args, thread_state=None):
def _MaintainSlicedDownloadTrackerFiles(src_obj_metadata, dst_url,
                                        download_file_name, logger,
                                        api_selector, num_components):
  def __init__(self, fp, tracker_file_name, src_obj_metadata, start_byte,
               end_byte):
  def mode(self):
  def write(self, data):  # pylint: disable=invalid-name
  def seek(self, offset, whence=os.SEEK_SET):  # pylint: disable=invalid-name
  def tell(self):  # pylint: disable=invalid-name
  def flush(self):  # pylint: disable=invalid-name
  def close(self):  # pylint: disable=invalid-name
def _PartitionObject(src_url,
                     src_obj_metadata,
                     dst_url,
                     download_file_name,
                     decryption_key=None):
def _DoSlicedDownload(src_url,
                      src_obj_metadata,
                      dst_url,
                      download_file_name,
                      command_obj,
                      logger,
                      copy_exception_handler,
                      api_selector,
                      decryption_key=None,
                      status_queue=None):
def _DownloadObjectToFileResumable(src_url,
                                   src_obj_metadata,
                                   dst_url,
                                   download_file_name,
                                   gsutil_api,
                                   logger,
                                   digesters,
                                   component_num=None,
                                   start_byte=0,
                                   end_byte=None,
                                   decryption_key=None):
def _DownloadObjectToFileNonResumable(src_url,
                                      src_obj_metadata,
                                      dst_url,
                                      download_file_name,
                                      gsutil_api,
                                      digesters,
                                      decryption_key=None):
def _DownloadObjectToFile(src_url,
                          src_obj_metadata,
                          dst_url,
                          gsutil_api,
                          logger,
                          command_obj,
                          copy_exception_handler,
                          allow_splitting=True,
                          decryption_key=None,
                          is_rsync=False,
                          preserve_posix=False,
                          use_stet=False):
def _ValidateAndCompleteDownload(logger,
                                 src_url,
                                 src_obj_metadata,
                                 dst_url,
                                 need_to_unzip,
                                 server_gzip,
                                 digesters,
                                 hash_algs,
                                 temporary_file_name,
                                 api_selector,
                                 bytes_transferred,
                                 gsutil_api,
                                 is_rsync=False,
                                 preserve_posix=False,
                                 use_stet=False):
def _CopyFileToFile(src_url, dst_url, status_queue=None, src_obj_metadata=None):
def _DummyTrackerCallback(_):
def _CopyObjToObjDaisyChainMode(src_url,
                                src_obj_metadata,
                                dst_url,
                                dst_obj_metadata,
                                preconditions,
                                gsutil_api,
                                logger,
                                decryption_key=None):
def GetSourceFieldsNeededForCopy(dst_is_cloud,
                                 skip_unsupported_objects,
                                 preserve_acl,
                                 is_rsync=False,
                                 preserve_posix=False,
                                 delete_source=False,
                                 file_size_will_change=False):
def WarnIfMvEarlyDeletionChargeApplies(src_url, src_obj_metadata, logger):
def MaybeSkipUnsupportedObject(src_url, src_obj_metadata):
def GetDecryptionCSEK(src_url, src_obj_metadata):
def PerformCopy(logger,
                src_url,
                dst_url,
                gsutil_api,
                command_obj,
                copy_exception_handler,
                src_obj_metadata=None,
                allow_splitting=True,
                headers=None,
                manifest=None,
                gzip_exts=None,
                is_rsync=False,
                preserve_posix=False,
                gzip_encoded=False,
                use_stet=False):
  def __init__(self, path):
  def _ParseManifest(self):
  def WasSuccessful(self, src):
  def _CreateManifestFile(self):
  def Set(self, url, key, value):
  def Initialize(self, source_url, destination_url):
  def SetResult(self, source_url, bytes_transferred, result, description=''):
  def _WriteRowToManifestFile(self, url):
  def _RemoveItemFromManifest(self, url):
  def __init__(self):
  def __init__(self):
def GetPathBeforeFinalDir(url, exp_src_url):
def ResolveWildcardsInPathBeforeFinalDir(src_url_path_sans_final_dir,
                                         exp_src_url):
def _GetPartitionInfo(file_size, max_components, default_component_size):
def _DeleteTempComponentObjectFn(cls, url_to_delete, thread_state=None):
def FilterExistingComponents(dst_args, existing_components, bucket_url,
                             gsutil_api): GID_ATTR = 'goog-reserved-posix-gid' MODE_ATTR = 'goog-reserved-posix-mode' NA_ID = -1 NA_MODE = -0o1 UID_ATTR = 'goog-reserved-posix-uid' def ParseAndSetPOSIXAttributes(path,
                               obj_metadata,
                               is_rsync=False,
                               preserve_posix=False):
  """Parses POSIX attributes from obj_metadata and sets them.

  Attributes will only be set if they exist in custom metadata. This function
  should only be called after ValidateFilePermissionAccess has been called for
  the specific file/object so as not to orphan files.

  Args:
    path: The local filesystem path for the file. Valid metadata attributes will
          be set for the file located at path, some attributes will only be set
          if preserve_posix is set to True.
    obj_metadata: The metadata for the file/object.
    is_rsync: Whether or not the caller is the rsync command. Used to determine
              if timeCreated should be used.
    preserve_posix: Whether or not all POSIX attributes should be set.
  """
  if obj_metadata is None:
    # This exception is meant for debugging purposes as it should never be
    # thrown unless there are unexpected code changes.
    raise CommandException('obj_metadata cannot be None for %s' % path)
  try:
    found_at, atime = GetValueFromObjectCustomMetadata(obj_metadata,
                                                       ATIME_ATTR,
                                                       default_value=NA_TIME)
    found_mt, mtime = GetValueFromObjectCustomMetadata(obj_metadata,
                                                       MTIME_ATTR,
                                                       default_value=NA_TIME)
    found_uid, uid = GetValueFromObjectCustomMetadata(obj_metadata,
                                                      UID_ATTR,
                                                      default_value=NA_ID)
    found_gid, gid = GetValueFromObjectCustomMetadata(obj_metadata,
                                                      GID_ATTR,
                                                      default_value=NA_ID)
    found_mode, mode = GetValueFromObjectCustomMetadata(obj_metadata,
                                                        MODE_ATTR,
                                                        default_value=NA_MODE)
    if not found_mode:
      mode = int(GetDefaultMode())
    if found_mt:
      mtime = long(mtime)
      if not preserve_posix:
        atime_tmp = os.stat(path).st_atime
        os.utime(path, (atime_tmp, mtime))
        return
    elif is_rsync:
      mtime = ConvertDatetimeToPOSIX(obj_metadata.timeCreated)
      os.utime(path, (mtime, mtime))
    if not preserve_posix:
      return
    if found_at:
      atime = long(atime)
    if atime > NA_TIME and mtime > NA_TIME:
      # Set both atime and mtime.
      os.utime(path, (atime, mtime))
    elif atime > NA_TIME and mtime <= NA_TIME:
      # atime is valid but mtime isn't.
      mtime_tmp = os.stat(path).st_mtime
      os.utime(path, (atime, mtime_tmp))
    elif atime <= NA_TIME and mtime > NA_TIME:
      # mtime is valid but atime isn't
      atime_tmp = os.stat(path).st_atime
      os.utime(path, (atime_tmp, mtime))
    if IS_WINDOWS:
      # Windows doesn't use POSIX uid/gid/mode unlike other systems. So there's
      # no point continuing.
      return
    # Only root can change ownership.
    if found_uid and os.geteuid() == 0:
      uid = int(uid)
    else:
      uid = NA_ID
    if found_gid:
      gid = int(gid)
    if uid > NA_ID and gid > NA_ID:
      # Set both uid and gid.
      os.chown(path, uid, gid)
    elif uid > NA_ID and gid <= NA_ID:
      # uid is valid but gid isn't.
      os.chown(path, uid, -1)
    elif uid <= NA_ID and gid > NA_ID:
      # gid is valid but uid isn't.
      os.chown(path, -1, gid)
    if found_mode:
      mode = int(str(mode), 8)
      os.chmod(path, mode)
  except ValueError:
    raise CommandException('Check POSIX attribute values for %s' %
                           obj_metadata.name) def ValidateFilePermissionAccess(url_str, uid=NA_ID, gid=NA_ID, mode=NA_MODE):
  """Validates that the user has file access if uid, gid, and mode are applied.

  Args:
    url_str: The path to the object for which this is validating.
    uid: A POSIX user ID.
    gid: A POSIX group ID.
    mode: A 3-digit, number representing POSIX permissions, must be in base-8.

  Returns:
    A (bool, str) tuple, True if and only if it's safe to copy the file, and a
    string containing details for the error.
  """
  # Windows doesn't use the POSIX system for file permissions, so all files will
  # validate.
  if IS_WINDOWS:
    return True, ''

  uid_present = uid > NA_ID
  gid_present = int(gid) > NA_ID
  mode_present = mode > NA_MODE
  # No need to perform validation if Posix attrs are not being preserved.
  if not (uid_present or gid_present or mode_present):
    return True, ''

  # The root user on non-Windows systems can access files regardless of their
  # permissions.
  if os.geteuid() == 0:
    return True, ''

  mode_valid = ValidatePOSIXMode(int(str(mode), 8))
  if mode_present:
    if not mode_valid:
      return False, 'Mode for %s won\'t allow read access.' % url_str
  else:
    # Calculate the default mode if the mode doesn't exist.
    # Convert mode to a 3-digit, base-8 integer.
    mode = int(GetDefaultMode())

  if uid_present:
    try:
      pwd.getpwuid(uid)
    except (KeyError, OverflowError):
      return (False, 'UID for %s doesn\'t exist on current system. uid: %d' %
              (url_str, uid))
  if gid_present:
    try:
      grp.getgrgid(gid)
    except (KeyError, OverflowError):
      return (False, 'GID for %s doesn\'t exist on current system. gid: %d' %
              (url_str, gid))

  # uid at this point must exist, but isn't necessarily the current user.
  # Likewise, gid must also exist at this point.
  uid_is_current_user = uid == os.getuid()

  # By this point uid and gid must exist on the system. However, the uid might
  # not match the current user's or the current user might not be a member of
  # the group identified by gid. In this case, the 'other' byte of the
  # permissions could provide sufficient access.
  mode = int(str(mode), 8)
  # Check that if the uid is not present and the gid and mode are, so that we
  # won't orphan the file. For example if the mode is set to 007, we can orphan
  # the file because the uid would default to the current user's ID and if the
  # current user wouldn't have read access or better, the file will be orphaned
  # even though they might otherwise have access through the gid or other bytes.
  if not uid_present and gid_present and mode_present and not bool(mode & U_R):
    return (False, 'Insufficient access with uid/gid/mode for %s, gid: %d, '
            'mode: %s' % (url_str, gid, oct(mode)[-3:]))
  if uid_is_current_user:
    valid = bool(mode & U_R)
    return (valid, '' if valid else
            'Insufficient access with uid/gid/mode for %s, uid: %d, '
            'mode: %s' % (url_str, uid, oct(mode)[-3:]))
  elif int(gid) in USER_GROUPS:
    valid = bool(mode & G_R)
    return (valid, '' if valid else
            'Insufficient access with uid/gid/mode for %s, gid: %d, '
            'mode: %s' % (url_str, gid, oct(mode)[-3:]))
  elif mode & O_R:
    return True, ''
  elif not uid_present and not gid_present and mode_valid:
    return True, ''
  return False, 'There was a problem validating %s.' % url_str def ValidatePOSIXMode(mode):
  """Validates whether the mode is valid.

  In order for the mode to be valid either the user, group, or other byte must
  be >= 4.

  Args:
    mode: The mode as a 3-digit, base-8 integer.

  Returns:
    True/False
  """
  return MODE_REGEX.match(oct(mode)[-3:]) and (mode & U_R or mode & G_R or
                                               mode & O_R)
import ast
import base64
import binascii
import datetime
import gzip
import logging
import os
import pickle
import pkgutil
import random
import re
import stat
import string
import sys
import threading
import boto
import gslib.tests.testcase as testcase
import six
from unittest import mock
from apitools.base.py import exceptions as apitools_exceptions
from boto import storage_uri
from boto.exception import ResumableTransferDisposition
from boto.exception import StorageResponseError
from boto.storage_uri import BucketStorageUri
from gslib import exception
from gslib import name_expansion
from gslib.cloud_api import ResumableUploadStartOverException
from gslib.commands.config import DEFAULT_SLICED_OBJECT_DOWNLOAD_THRESHOLD
from gslib.cs_api_map import ApiSelector
from gslib.daisy_chain_wrapper import _DEFAULT_DOWNLOAD_CHUNK_SIZE
from gslib.discard_messages_queue import DiscardMessagesQueue
from gslib.exception import InvalidUrlError
from gslib.gcs_json_api import GcsJsonApi
from gslib.parallel_tracker_file import ObjectFromTracker
from gslib.parallel_tracker_file import WriteParallelUploadTrackerFile
from gslib.project_id import PopulateProjectId
from gslib.storage_url import StorageUrlFromString
from gslib.tests.rewrite_helper import EnsureRewriteResumeCallbackHandler
from gslib.tests.rewrite_helper import HaltingRewriteCallbackHandler
from gslib.tests.rewrite_helper import RewriteHaltException
from gslib.tests.testcase.base import NotParallelizable
from gslib.tests.testcase.integration_testcase import SkipForGS
from gslib.tests.testcase.integration_testcase import SkipForS3
from gslib.tests.testcase.integration_testcase import SkipForXML
from gslib.tests.testcase.integration_testcase import SkipForJSON
from gslib.tests.util import BuildErrorRegex
from gslib.tests.util import GenerationFromURI as urigen
from gslib.tests.util import HaltingCopyCallbackHandler
from gslib.tests.util import HaltOneComponentCopyCallbackHandler
from gslib.tests.util import HAS_GS_PORT
from gslib.tests.util import HAS_S3_CREDS
from gslib.tests.util import ObjectToURI as suri
from gslib.tests.util import ORPHANED_FILE
from gslib.tests.util import POSIX_GID_ERROR
from gslib.tests.util import POSIX_INSUFFICIENT_ACCESS_ERROR
from gslib.tests.util import POSIX_MODE_ERROR
from gslib.tests.util import POSIX_UID_ERROR
from gslib.tests.util import SequentialAndParallelTransfer
from gslib.tests.util import SetBotoConfigForTest
from gslib.tests.util import SetEnvironmentForTest
from gslib.tests.util import TailSet
from gslib.tests.util import TEST_ENCRYPTION_KEY1
from gslib.tests.util import TEST_ENCRYPTION_KEY1_SHA256_B64
from gslib.tests.util import TEST_ENCRYPTION_KEY2
from gslib.tests.util import TEST_ENCRYPTION_KEY3
from gslib.tests.util import unittest
from gslib.third_party.storage_apitools import storage_v1_messages as apitools_messages
from gslib.tracker_file import DeleteTrackerFile
from gslib.tracker_file import GetRewriteTrackerFilePath
from gslib.tracker_file import GetSlicedDownloadTrackerFilePaths
from gslib.ui_controller import BytesToFixedWidthString
from gslib.utils import hashing_helper
from gslib.utils.boto_util import UsingCrcmodExtension
from gslib.utils.constants import START_CALLBACK_PER_BYTES
from gslib.utils.constants import UTF8
from gslib.utils.copy_helper import GetTrackerFilePath
from gslib.utils.copy_helper import PARALLEL_UPLOAD_STATIC_SALT
from gslib.utils.copy_helper import PARALLEL_UPLOAD_TEMP_NAMESPACE
from gslib.utils.copy_helper import TrackerFileType
from gslib.utils.hashing_helper import CalculateB64EncodedMd5FromContents
from gslib.utils.hashing_helper import CalculateMd5FromContents
from gslib.utils.hashing_helper import GetMd5
from gslib.utils.metadata_util import CreateCustomMetadata
from gslib.utils.posix_util import GID_ATTR
from gslib.utils.posix_util import MODE_ATTR
from gslib.utils.posix_util import NA_ID
from gslib.utils.posix_util import NA_MODE
from gslib.utils.posix_util import UID_ATTR
from gslib.utils.posix_util import ParseAndSetPOSIXAttributes
from gslib.utils.posix_util import ValidateFilePermissionAccess
from gslib.utils.posix_util import ValidatePOSIXMode
from gslib.utils.retry_util import Retry
from gslib.utils.system_util import IS_WINDOWS
from gslib.utils.text_util import get_random_ascii_chars
from gslib.utils.unit_util import EIGHT_MIB
from gslib.utils.unit_util import HumanReadableToBytes
from gslib.utils.unit_util import MakeHumanReadable
from gslib.utils.unit_util import ONE_KIB
from gslib.utils.unit_util import ONE_MIB
from six.moves import http_client
from six.moves import range
from six.moves import xrange
  from gslib.tests import util
  from gslib.tests.util import DEFAULT_MODE
  from gslib.tests.util import GetInvalidGid
  from gslib.tests.util import GetNonPrimaryGid
  from gslib.tests.util import GetPrimaryGid
  from gslib.tests.util import INVALID_UID
  from gslib.tests.util import USER_ID
    self.RunGsUtil(['stat', '%s/obj1' % suri(bucket_uri)])

  def test_cp_minus_m_streaming_upload(self):
    """Tests that cp -m - anything is disallowed."""
    stderr = self.RunGsUtil(['-m', 'cp', '-', 'file'],
                            return_stderr=True,
                            expected_status=1)
    self.assertIn(
        'CommandException: Cannot upload from a stream when using gsutil -m',
        stderr)

  @SequentialAndParallelTransfer
  def test_cp_overwrites_existing_destination(self):
    key_uri = self.CreateObject(contents=b'foo')
    fpath = self.CreateTempFile(contents=b'bar')
    stderr = self.RunGsUtil(['cp', suri(key_uri), fpath], return_stderr=True)
    with open(fpath, 'rb') as f:
      self.assertEqual(f.read(), b'foo')

  @SequentialAndParallelTransfer
  def test_downloads_are_reliable_with_more_than_one_gsutil_instance(self):
    test_file_count = 10
    temporary_directory = self.CreateTempDir()
    bucket_uri = self.CreateBucket(test_objects=test_file_count)

    cp_args = ['cp', suri(bucket_uri, '*'), temporary_directory]
    threads = []
    for _ in range(2):
      thread = threading.Thread(target=self.RunGsUtil, args=[cp_args])
      thread.start()
      threads.append(thread)
    [t.join() for t in threads]

    self.assertEqual(len(os.listdir(temporary_directory)), test_file_count)


class TestCpUnitTests(testcase.GsUtilUnitTestCase):
  """Unit tests for gsutil cp."""

  def testDownloadWithNoHashAvailable(self):
    """Tests a download with no valid server-supplied hash."""
    # S3 should have a special message for non-MD5 etags.
    bucket_uri = self.CreateBucket(provider='s3')
    object_uri = self.CreateObject(bucket_uri=bucket_uri, contents=b'foo')
    object_uri.get_key().etag = '12345'  # Not an MD5
    dst_dir = self.CreateTempDir()

    log_handler = self.RunCommand('cp', [suri(object_uri), dst_dir],
                                  return_log_handler=True)
    warning_messages = log_handler.messages['warning']
    self.assertEquals(2, len(warning_messages))
    self.assertRegex(
        warning_messages[0], r'Non-MD5 etag \(12345\) present for key .*, '
        r'data integrity checks are not possible')
    self.assertIn('Integrity cannot be assured', warning_messages[1])

  def testDownloadWithDestinationEndingWithDelimiterRaisesError(self):
    """Tests a download with no valid server-supplied hash."""
    # S3 should have a special message for non-MD5 etags.
    bucket_uri = self.CreateBucket(provider='s3')
    object_uri = self.CreateObject(bucket_uri=bucket_uri, contents=b'foo')
    destination_path = 'random_dir' + os.path.sep

    with self.assertRaises(InvalidUrlError) as error:
      self.RunCommand('cp', [suri(object_uri), destination_path])
      self.assertEqual(str(error), 'Invalid destination path: random_dir/')

  def test_object_and_prefix_same_name(self):
    bucket_uri = self.CreateBucket()
    object_uri = self.CreateObject(bucket_uri=bucket_uri,
                                   object_name='foo',
                                   contents=b'foo')
    self.CreateObject(bucket_uri=bucket_uri,
                      object_name='foo/bar',
                      contents=b'bar')
    fpath = self.CreateTempFile()
    # MockKey doesn't support hash_algs, so the MD5 will not match.
    with SetBotoConfigForTest([('GSUtil', 'check_hashes', 'never')]):
      self.RunCommand('cp', [suri(object_uri), fpath])
    with open(fpath, 'rb') as f:
      self.assertEqual(f.read(), b'foo')

  def test_cp_upload_respects_no_hashes(self):
    bucket_uri = self.CreateBucket()
    fpath = self.CreateTempFile(contents=b'abcd')
    with SetBotoConfigForTest([('GSUtil', 'check_hashes', 'never')]):
      log_handler = self.RunCommand('cp', [fpath, suri(bucket_uri)],
                                    return_log_handler=True)
    warning_messages = log_handler.messages['warning']
    self.assertEquals(1, len(warning_messages))
    self.assertIn('Found no hashes to validate object upload',
                  warning_messages[0])

  def test_shim_translates_flags(self):
    bucket_uri = self.CreateBucket()
    fpath = self.CreateTempFile(contents=b'abcd')
    with SetBotoConfigForTest([('GSUtil', 'use_gcloud_storage', 'True'),
                               ('GSUtil', 'hidden_shim_mode', 'dry_run')]):
      with SetEnvironmentForTest({
          'CLOUDSDK_CORE_PASS_CREDENTIALS_TO_GSUTIL': 'True',
          'CLOUDSDK_ROOT_DIR': 'fake_dir',
      }):
        mock_log_handler = self.RunCommand(
            'cp',
            ['-r', '-R', '-e', fpath, suri(bucket_uri)],
            return_log_handler=True)
        info_lines = '\n'.join(mock_log_handler.messages['info'])
        self.assertIn(
            'Gcloud Storage Command: {} alpha storage cp'
            ' -r -r --ignore-symlinks {} {}'.format(
                os.path.join('fake_dir', 'bin', 'gcloud'), fpath,
                suri(bucket_uri)), info_lines)

  @unittest.skipIf(IS_WINDOWS, 'POSIX attributes not available on Windows.')
  @mock.patch('os.geteuid', new=mock.Mock(return_value=0))
  @mock.patch.object(os, 'chown', autospec=True)
  def test_posix_runs_chown_as_super_user(self, mock_chown):
    fpath = self.CreateTempFile(contents=b'abcd')
    obj = apitools_messages.Object()
    obj.metadata = CreateCustomMetadata(entries={UID_ATTR: USER_ID})

    ParseAndSetPOSIXAttributes(fpath, obj, False, True)