gsutil/gslib/aclhelpers.py

# -*- 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)