<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> from gslib import aclhelpers <EOL> from gslib . cloud_api import AccessDeniedException <EOL> from gslib . cloud_api import BadRequestException <EOL> from gslib . cloud_api import Preconditions <EOL> from gslib . cloud_api import ServiceException <EOL> from gslib . command import Command <EOL> from gslib . command import SetAclExceptionHandler <EOL> from gslib . command import SetAclFuncWrapper <EOL> from gslib . command_argument import CommandArgument <EOL> from gslib . cs_api_map import ApiSelector <EOL> from gslib . exception import CommandException <EOL> from gslib . help_provider import CreateHelpText <EOL> from gslib . storage_url import StorageUrlFromString <EOL> from gslib . third_party . storage_apitools import storage_v1_messages as apitools_messages <EOL> from gslib . util import NO_MAX <EOL> from gslib . util import Retry <EOL> from gslib . util import UrlsAreForSingleProvider <EOL> _SET_SYNOPSIS = """<STR_LIT>""" <EOL> _GET_SYNOPSIS = """<STR_LIT>""" <EOL> _CH_SYNOPSIS = """<STR_LIT>""" <EOL> _GET_DESCRIPTION = """<STR_LIT>""" <EOL> _SET_DESCRIPTION = """<STR_LIT>""" <EOL> _CH_DESCRIPTION = """<STR_LIT>""" <EOL> _SYNOPSIS = ( _SET_SYNOPSIS + _GET_SYNOPSIS . lstrip ( '<STR_LIT:\n>' ) + <EOL> _CH_SYNOPSIS . lstrip ( '<STR_LIT:\n>' ) + '<STR_LIT>' ) <EOL> _DESCRIPTION = ( """<STR_LIT>""" + '<STR_LIT:\n>' . join ( [ _GET_DESCRIPTION , _SET_DESCRIPTION , _CH_DESCRIPTION ] ) ) <EOL> _DETAILED_HELP_TEXT = CreateHelpText ( _SYNOPSIS , _DESCRIPTION ) <EOL> _get_help_text = CreateHelpText ( _GET_SYNOPSIS , _GET_DESCRIPTION ) <EOL> _set_help_text = CreateHelpText ( _SET_SYNOPSIS , _SET_DESCRIPTION ) <EOL> _ch_help_text = CreateHelpText ( _CH_SYNOPSIS , _CH_DESCRIPTION ) <EOL> def _ApplyExceptionHandler ( cls , exception ) : <EOL> cls . logger . error ( '<STR_LIT>' , exception ) <EOL> cls . everything_set_okay = False <EOL> def _ApplyAclChangesWrapper ( cls , url_or_expansion_result , thread_state = None ) : <EOL> cls . ApplyAclChanges ( url_or_expansion_result , thread_state = thread_state ) <EOL> class AclCommand ( Command ) : <EOL> """<STR_LIT>""" <EOL> command_spec = Command . CreateCommandSpec ( <EOL> '<STR_LIT>' , <EOL> command_name_aliases = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> usage_synopsis = _SYNOPSIS , <EOL> min_args = <NUM_LIT:2> , <EOL> max_args = NO_MAX , <EOL> supported_sub_args = '<STR_LIT>' , <EOL> file_url_ok = False , <EOL> provider_url_ok = False , <EOL> urls_start_arg = <NUM_LIT:1> , <EOL> gs_api_support = [ ApiSelector . XML , ApiSelector . JSON ] , <EOL> gs_default_api = ApiSelector . JSON , <EOL> argparse_arguments = { <EOL> '<STR_LIT>' : [ <EOL> CommandArgument . MakeFileURLOrCannedACLArgument ( ) , <EOL> CommandArgument . MakeZeroOrMoreCloudURLsArgument ( ) <EOL> ] , <EOL> '<STR_LIT>' : [ <EOL> CommandArgument . MakeNCloudURLsArgument ( <NUM_LIT:1> ) <EOL> ] , <EOL> '<STR_LIT>' : [ <EOL> CommandArgument . MakeZeroOrMoreCloudURLsArgument ( ) <EOL> ] , <EOL> } <EOL> ) <EOL> help_spec = Command . HelpSpec ( <EOL> help_name = '<STR_LIT>' , <EOL> help_name_aliases = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] , <EOL> help_type = '<STR_LIT>' , <EOL> help_one_line_summary = '<STR_LIT>' , <EOL> help_text = _DETAILED_HELP_TEXT , <EOL> subcommand_help_text = { <EOL> '<STR_LIT>' : _get_help_text , '<STR_LIT>' : _set_help_text , '<STR_LIT>' : _ch_help_text } , <EOL> ) <EOL> def _CalculateUrlsStartArg ( self ) : <EOL> if not self . args : <EOL> self . RaiseWrongNumberOfArgumentsException ( ) <EOL> if ( self . args [ <NUM_LIT:0> ] . lower ( ) == '<STR_LIT>' ) or ( self . command_alias_used == '<STR_LIT>' ) : <EOL> return <NUM_LIT:1> <EOL> else : <EOL> return <NUM_LIT:0> <EOL> def _SetAcl ( self ) : <EOL> """<STR_LIT>""" <EOL> self . continue_on_error = False <EOL> if self . sub_opts : <EOL> for o , unused_a in self . sub_opts : <EOL> if o == '<STR_LIT>' : <EOL> self . all_versions = True <EOL> elif o == '<STR_LIT>' : <EOL> self . continue_on_error = True <EOL> elif o == '<STR_LIT>' or o == '<STR_LIT>' : <EOL> self . recursion_requested = True <EOL> else : <EOL> self . RaiseInvalidArgumentException ( ) <EOL> try : <EOL> self . SetAclCommandHelper ( SetAclFuncWrapper , SetAclExceptionHandler ) <EOL> except AccessDeniedException , unused_e : <EOL> self . _WarnServiceAccounts ( ) <EOL> raise <EOL> if not self . everything_set_okay : <EOL> raise CommandException ( '<STR_LIT>' ) <EOL> def _ChAcl ( self ) : <EOL> """<STR_LIT>""" <EOL> self . parse_versions = True <EOL> self . changes = [ ] <EOL> self . continue_on_error = False <EOL> if self . sub_opts : <EOL> for o , a in self . sub_opts : <EOL> if o == '<STR_LIT>' : <EOL> self . continue_on_error = True <EOL> elif o == '<STR_LIT>' : <EOL> if '<STR_LIT>' in a : <EOL> raise CommandException ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> self . changes . append ( <EOL> aclhelpers . AclChange ( a , scope_type = aclhelpers . ChangeType . GROUP ) ) <EOL> elif o == '<STR_LIT>' : <EOL> self . changes . append ( <EOL> aclhelpers . AclChange ( a , scope_type = aclhelpers . ChangeType . PROJECT ) ) <EOL> elif o == '<STR_LIT>' : <EOL> self . changes . append ( <EOL> aclhelpers . AclChange ( a , scope_type = aclhelpers . ChangeType . USER ) ) <EOL> elif o == '<STR_LIT>' : <EOL> self . changes . append ( aclhelpers . AclDel ( a ) ) <EOL> elif o == '<STR_LIT>' or o == '<STR_LIT>' : <EOL> self . recursion_requested = True <EOL> else : <EOL> self . RaiseInvalidArgumentException ( ) <EOL> if not self . changes : <EOL> raise CommandException ( <EOL> '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> if ( not UrlsAreForSingleProvider ( self . args ) or <EOL> StorageUrlFromString ( self . args [ <NUM_LIT:0> ] ) . scheme != '<STR_LIT>' ) : <EOL> raise CommandException ( <EOL> '<STR_LIT>' . format ( <EOL> self . command_name ) ) <EOL> self . everything_set_okay = True <EOL> self . ApplyAclFunc ( _ApplyAclChangesWrapper , _ApplyExceptionHandler , <EOL> self . args ) <EOL> if not self . everything_set_okay : <EOL> raise CommandException ( '<STR_LIT>' ) <EOL> def _RaiseForAccessDenied ( self , url ) : <EOL> self . _WarnServiceAccounts ( ) <EOL> raise CommandException ( '<STR_LIT>' <EOL> '<STR_LIT>' % url ) <EOL> @ Retry ( ServiceException , tries = <NUM_LIT:3> , timeout_secs = <NUM_LIT:1> ) <EOL> def ApplyAclChanges ( self , name_expansion_result , thread_state = None ) : <EOL> """<STR_LIT>""" <EOL> if thread_state : <EOL> gsutil_api = thread_state <EOL> else : <EOL> gsutil_api = self . gsutil_api <EOL> url = name_expansion_result . expanded_storage_url <EOL> if url . IsBucket ( ) : <EOL> bucket = gsutil_api . GetBucket ( url . bucket_name , provider = url . scheme , <EOL> fields = [ '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> current_acl = bucket . acl <EOL> elif url . IsObject ( ) : <EOL> gcs_object = gsutil_api . GetObjectMetadata ( <EOL> url . bucket_name , url . object_name , provider = url . scheme , <EOL> generation = url . generation , <EOL> fields = [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] ) <EOL> current_acl = gcs_object . acl <EOL> if not current_acl : <EOL> self . _RaiseForAccessDenied ( url ) <EOL> modification_count = <NUM_LIT:0> <EOL> for change in self . changes : <EOL> modification_count += change . Execute ( url , current_acl , '<STR_LIT>' , self . logger ) <EOL> if modification_count == <NUM_LIT:0> : <EOL> self . logger . info ( '<STR_LIT>' , url ) <EOL> return <EOL> try : <EOL> if url . IsBucket ( ) : <EOL> preconditions = Preconditions ( meta_gen_match = bucket . metageneration ) <EOL> bucket_metadata = apitools_messages . Bucket ( acl = current_acl ) <EOL> gsutil_api . PatchBucket ( url . bucket_name , bucket_metadata , <EOL> preconditions = preconditions , <EOL> provider = url . scheme , fields = [ '<STR_LIT:id>' ] ) <EOL> else : <EOL> preconditions = Preconditions ( gen_match = gcs_object . generation , <EOL> meta_gen_match = gcs_object . metageneration ) <EOL> object_metadata = apitools_messages . Object ( acl = current_acl ) <EOL> gsutil_api . PatchObjectMetadata ( <EOL> url . bucket_name , url . object_name , object_metadata , <EOL> preconditions = preconditions , provider = url . scheme , <EOL> generation = url . generation ) <EOL> except BadRequestException as e : <EOL> raise CommandException ( '<STR_LIT>' % str ( e ) ) <EOL> except AccessDeniedException : <EOL> self . _RaiseForAccessDenied ( url ) <EOL> self . logger . info ( '<STR_LIT>' , url ) <EOL> def RunCommand ( self ) : <EOL> """<STR_LIT>""" <EOL> action_subcommand = self . args . pop ( <NUM_LIT:0> ) <EOL> self . ParseSubOpts ( check_args = True ) <EOL> self . def_acl = False <EOL> if action_subcommand == '<STR_LIT>' : <EOL> self . GetAndPrintAcl ( self . args [ <NUM_LIT:0> ] ) <EOL> elif action_subcommand == '<STR_LIT>' : <EOL> self . _SetAcl ( ) <EOL> elif action_subcommand in ( '<STR_LIT>' , '<STR_LIT>' ) : <EOL> self . _ChAcl ( ) <EOL> else : <EOL> raise CommandException ( ( '<STR_LIT>' <EOL> '<STR_LIT>' ) % <EOL> ( action_subcommand , self . command_name ) ) <EOL> return <NUM_LIT:0> </s>
