# -*- coding: utf-8 -*-
# Copyright 2013 Google Inc. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""Contains helper objects for changing and deleting ACLs."""
from __future__ import absolute_import
import re
from gslib.exception import CommandException
from gslib.third_party.storage_apitools import storage_v1_messages as apitools_messages
class ChangeType(object):
USER = 'User'
GROUP = 'Group'
PROJECT = 'Project'
class AclChange(object):
"""Represents a logical change to an access control list."""
public_scopes = ['AllAuthenticatedUsers', 'AllUsers']
id_scopes = ['UserById', 'GroupById']
email_scopes = ['UserByEmail', 'GroupByEmail']
domain_scopes = ['GroupByDomain']
project_scopes = ['Project']
scope_types = (public_scopes + id_scopes + email_scopes + domain_scopes
+ project_scopes)
public_entity_all_users = 'allUsers'
public_entity_all_auth_users = 'allAuthenticatedUsers'
public_entity_types = (public_entity_all_users, public_entity_all_auth_users)
project_entity_prefixes = ('project-editors-', 'project-owners-',
'project-viewers-')
group_entity_prefix = 'group-'
user_entity_prefix = 'user-'
domain_entity_prefix = 'domain-'
project_entity_prefix = 'project-'
permission_shorthand_mapping = {
'R': 'READER',
'W': 'WRITER',
'FC': 'OWNER',
'O': 'OWNER',
'READ': 'READER',
'WRITE': 'WRITER',
'FULL_CONTROL': 'OWNER'
}
def __init__(self, acl_change_descriptor, scope_type):
"""Creates an AclChange object.
Args:
acl_change_descriptor: An acl change as described in the "ch" section of
the "acl" command's help.
scope_type: Either ChangeType.USER or ChangeType.GROUP or
ChangeType.PROJECT, specifying the extent of the scope.
"""
self.identifier = ''
self.raw_descriptor = acl_change_descriptor
self._Parse(acl_change_descriptor, scope_type)
self._Validate()
def __str__(self):
return 'AclChange<{0}|{1}|{2}>'.format(
self.scope_type, self.perm, self.identifier)
def _Parse(self, change_descriptor, scope_type):
"""Parses an ACL Change descriptor."""
def _ClassifyScopeIdentifier(text):
re_map = {
'AllAuthenticatedUsers': r'^(AllAuthenticatedUsers|AllAuth)$',
'AllUsers': '^(AllUsers|All)$',
'Email': r'^.+@.+\..+$',
'Id': r'^[0-9A-Fa-f]{64}$',
'Domain': r'^[^@]+\.[^@]+$',
'Project': r'(owners|editors|viewers)\-.+$',
}
for type_string, regex in re_map.items():
if re.match(regex, text, re.IGNORECASE):
return type_string
if change_descriptor.count(':') != 1:
raise CommandException('{0} is an invalid change description.'
.format(change_descriptor))
scope_string, perm_token = change_descriptor.split(':')
perm_token = perm_token.upper()
if perm_token in self.permission_shorthand_mapping:
self.perm = self.permission_shorthand_mapping[perm_token]
else:
self.perm = perm_token
scope_class = _ClassifyScopeIdentifier(scope_string)
if scope_class == 'Domain':
# This may produce an invalid UserByDomain scope,
# which is good because then validate can complain.
self.scope_type = '{0}ByDomain'.format(scope_type)
self.identifier = scope_string
elif scope_class in ('Email', 'Id'):
self.scope_type = '{0}By{1}'.format(scope_type, scope_class)
self.identifier = scope_string
elif scope_class == 'AllAuthenticatedUsers':
self.scope_type = 'AllAuthenticatedUsers'
elif scope_class == 'AllUsers':
self.scope_type = 'AllUsers'
elif scope_class == 'Project':
self.scope_type = 'Project'
self.identifier = scope_string
else:
# This is just a fallback, so we set it to something
# and the validate step has something to go on.
self.scope_type = scope_string
def _Validate(self):
"""Validates a parsed AclChange object."""
def _ThrowError(msg):
raise CommandException('{0} is not a valid ACL change\n{1}'
.format(self.raw_descriptor, msg))
if self.scope_type not in self.scope_types:
_ThrowError('{0} is not a valid scope type'.format(self.scope_type))
if self.scope_type in self.public_scopes and self.identifier:
_ThrowError('{0} requires no arguments'.format(self.scope_type))
if self.scope_type in self.id_scopes and not self.identifier:
_ThrowError('{0} requires an id'.format(self.scope_type))
if self.scope_type in self.email_scopes and not self.identifier:
_ThrowError('{0} requires an email address'.format(self.scope_type))
if self.scope_type in self.domain_scopes and not self.identifier:
_ThrowError('{0} requires domain'.format(self.scope_type))
if self.perm not in self.permission_shorthand_mapping.values():
perms = ', '.join(self.permission_shorthand_mapping.values())
_ThrowError('Allowed permissions are {0}'.format(perms))
def _YieldMatchingEntries(self, current_acl):
"""Generator that yields entries that match the change descriptor.
Args:
current_acl: A list of apitools_messages.BucketAccessControls or
ObjectAccessControls which will be searched for matching
entries.
Yields:
An apitools_messages.BucketAccessControl or ObjectAccessControl.
"""
for entry in current_acl:
if (self.scope_type in ('UserById', 'GroupById') and
entry.entityId and self.identifier == entry.entityId):
yield entry
elif (self.scope_type in ('UserByEmail', 'GroupByEmail')
and entry.email and self.identifier == entry.email):
yield entry
elif (self.scope_type == 'GroupByDomain' and
entry.domain and self.identifier == entry.domain):
yield entry
elif (self.scope_type == 'Project' and
entry.domain and self.identifier == entry.project):
yield entry
elif (self.scope_type == 'AllUsers' and
entry.entity.lower() == self.public_entity_all_users.lower()):
yield entry
elif (self.scope_type == 'AllAuthenticatedUsers' and
entry.entity.lower() == self.public_entity_all_auth_users.lower()):
yield entry
def _AddEntry(self, current_acl, entry_class):
"""Adds an entry to current_acl."""
if self.scope_type == 'UserById':
entry = entry_class(entityId=self.identifier, role=self.perm,
entity=self.user_entity_prefix + self.identifier)
elif self.scope_type == 'GroupById':
entry = entry_class(entityId=self.identifier, role=self.perm,
entity=self.group_entity_prefix + self.identifier)
elif self.scope_type == 'Project':
entry = entry_class(entityId=self.identifier, role=self.perm,
entity=self.project_entity_prefix + self.identifier)
elif self.scope_type == 'UserByEmail':
entry = entry_class(email=self.identifier, role=self.perm,
entity=self.user_entity_prefix + self.identifier)
elif self.scope_type == 'GroupByEmail':
entry = entry_class(email=self.identifier, role=self.perm,
entity=self.group_entity_prefix + self.identifier)
elif self.scope_type == 'GroupByDomain':
entry = entry_class(domain=self.identifier, role=self.perm,
entity=self.domain_entity_prefix + self.identifier)
elif self.scope_type == 'AllAuthenticatedUsers':
entry = entry_class(entity=self.public_entity_all_auth_users,
role=self.perm)
elif self.scope_type == 'AllUsers':
entry = entry_class(entity=self.public_entity_all_users, role=self.perm)
else:
raise CommandException('Add entry to ACL got unexpected scope type %s.' %
self.scope_type)
current_acl.append(entry)
def _GetEntriesClass(self, current_acl):
# Entries will share the same class, so just return the first one.
for acl_entry in current_acl:
return acl_entry.__class__
# It's possible that a default object ACL is empty, so if we have
# an empty list, assume it is an object ACL.
return apitools_messages.ObjectAccessControl().__class__
def Execute(self, storage_url, current_acl, command_name, logger):
"""Executes the described change on an ACL.
Args:
storage_url: StorageUrl representing the object to change.
current_acl: A list of ObjectAccessControls or
BucketAccessControls to permute.
command_name: String name of comamnd being run (e.g., 'acl').
logger: An instance of logging.Logger.
Returns:
The number of changes that were made.
"""
logger.debug(
'Executing %s %s on %s', command_name, self.raw_descriptor, storage_url)
if self.perm == 'WRITER':
if command_name == 'acl' and storage_url.IsObject():
logger.warning(
'Skipping %s on %s, as WRITER does not apply to objects',
self.raw_descriptor, storage_url)
return 0
elif command_name == 'defacl':
raise CommandException('WRITER cannot be set as a default object ACL '
'because WRITER does not apply to objects')
entry_class = self._GetEntriesClass(current_acl)
matching_entries = list(self._YieldMatchingEntries(current_acl))
change_count = 0
if matching_entries:
for entry in matching_entries:
if entry.role != self.perm:
entry.role = self.perm
change_count += 1
else:
self._AddEntry(current_acl, entry_class)
change_count = 1
logger.debug('New Acl:\n%s', str(current_acl))
return change_count
class AclDel(object):
"""Represents a logical change from an access control list."""
scope_regexes = {
r'All(Users)?$': 'AllUsers',
r'AllAuth(enticatedUsers)?$': 'AllAuthenticatedUsers',
}
def __init__(self, identifier):
self.raw_descriptor = '-d {0}'.format(identifier)
self.identifier = identifier
for regex, scope in self.scope_regexes.items():
if re.match(regex, self.identifier, re.IGNORECASE):
self.identifier = scope
self.scope_type = 'Any'
self.perm = 'NONE'
def _YieldMatchingEntries(self, current_acl):
"""Generator that yields entries that match the change descriptor.
Args:
current_acl: An instance of apitools_messages.BucketAccessControls or
ObjectAccessControls which will be searched for matching
entries.
Yields:
An apitools_messages.BucketAccessControl or ObjectAccessControl.
"""
for entry in current_acl:
if entry.entityId and self.identifier == entry.entityId:
yield entry
elif entry.email and self.identifier == entry.email:
yield entry
elif entry.domain and self.identifier == entry.domain:
yield entry
elif entry.projectTeam:
project_team = entry.projectTeam
acl_label = project_team.team + '-' + project_team.projectNumber
if acl_label == self.identifier:
yield entry
elif entry.entity.lower() == 'allusers' and self.identifier == 'AllUsers':
yield entry
elif (entry.entity.lower() == 'allauthenticatedusers' and
self.identifier == 'AllAuthenticatedUsers'):
yield entry
def Execute(self, storage_url, current_acl, command_name, logger):
logger.debug(
'Executing %s %s on %s', command_name, self.raw_descriptor, storage_url)
matching_entries = list(self._YieldMatchingEntries(current_acl))
for entry in matching_entries:
current_acl.remove(entry)
logger.debug('New Acl:\n%s', str(current_acl))
return len(matching_entries)