path: root/src/mailman/rest/users.py

# Copyright (C) 2011-2014 by the Free Software Foundation, Inc.
#
# This file is part of GNU Mailman.
#
# GNU Mailman is free software: you can redistribute it and/or modify it under
# the terms of the GNU General Public License as published by the Free
# Software Foundation, either version 3 of the License, or (at your option)
# any later version.
#
# GNU Mailman is distributed in the hope that it will be useful, but WITHOUT
# ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
# FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License for
# more details.
#
# You should have received a copy of the GNU General Public License along with
# GNU Mailman.  If not, see <http://www.gnu.org/licenses/>.

"""REST for users."""

__all__ = [
    'AUser',
    'AddressUser',
    'AllUsers',
    'Login',
    ]


from lazr.config import as_boolean
from mailman.config import config
from mailman.core.errors import (
    ReadOnlyPATCHRequestError, UnknownPATCHRequestError)
from mailman.interfaces.address import ExistingAddressError
from mailman.interfaces.usermanager import IUserManager
from mailman.rest.addresses import UserAddresses
from mailman.rest.helpers import (
    BadRequest, CollectionMixin, GetterSetter, NotFound, bad_request, child,
    conflict, created, etag, forbidden, no_content, not_found, okay, paginate,
    path_to)
from mailman.rest.preferences import Preferences
from mailman.rest.validator import PatchValidator, Validator
from passlib.utils import generate_password as generate
from uuid import UUID
from zope.component import getUtility



# Attributes of a user which can be changed via the REST API.
class PasswordEncrypterGetterSetter(GetterSetter):
    def __init__(self):
        super(PasswordEncrypterGetterSetter, self).__init__(
            config.password_context.encrypt)
    def get(self, obj, attribute):
        assert attribute == 'cleartext_password'
        super(PasswordEncrypterGetterSetter, self).get(obj, 'password')
    def put(self, obj, attribute, value):
        assert attribute == 'cleartext_password'
        super(PasswordEncrypterGetterSetter, self).put(obj, 'password', value)


ATTRIBUTES = dict(
    display_name=GetterSetter(str),
    cleartext_password=PasswordEncrypterGetterSetter(),
    )


CREATION_FIELDS = dict(
    email=str,
    display_name=str,
    password=str,
    _optional=('display_name', 'password'),
    )



def create_user(arguments, response):
    """Create a new user."""
    # We can't pass the 'password' argument to the user creation method, so
    # strip that out (if it exists), then create the user, adding the password
    # after the fact if successful.
    password = arguments.pop('password', None)
    try:
        user = getUtility(IUserManager).create_user(**arguments)
    except ExistingAddressError as error:
        bad_request(
            response, 'Address already exists: {}'.format(error.address))
        return None
    if password is None:
        # This will have to be reset since it cannot be retrieved.
        password = generate(int(config.passwords.password_length))
    user.password = config.password_context.encrypt(password)
    location = path_to('users/{}'.format(user.user_id.int))
    created(response, location)
    return user



class _UserBase(CollectionMixin):
    """Shared base class for user representations."""

    def _resource_as_dict(self, user):
        """See `CollectionMixin`."""
        # The canonical URL for a user is their unique user id, although we
        # can always look up a user based on any registered and validated
        # email address associated with their account.  The user id is a UUID,
        # but we serialize its integer equivalent.
        user_id = user.user_id.int
        resource = dict(
            user_id=user_id,
            created_on=user.created_on,
            self_link=path_to('users/{}'.format(user_id)),
            )
        # Add the password attribute, only if the user has a password.  Same
        # with the real name.  These could be None or the empty string.
        if user.password:
            resource['password'] = user.password
        if user.display_name:
            resource['display_name'] = user.display_name
        return resource

    @paginate
    def _get_collection(self, request):
        """See `CollectionMixin`."""
        return list(getUtility(IUserManager).users)



class AllUsers(_UserBase):
    """The users."""

    def on_get(self, request, response):
        """/users"""
        resource = self._make_collection(request)
        okay(response, etag(resource))

    def on_post(self, request, response):
        """Create a new user."""
        try:
            validator = Validator(**CREATION_FIELDS)
            arguments = validator(request)
        except ValueError as error:
            bad_request(response, str(error))
            return
        create_user(arguments, response)



class AUser(_UserBase):
    """A user."""

    def __init__(self, user_identifier):
        """Get a user by various type of identifiers.

        :param user_identifier: The identifier used to retrieve the user.  The
            identifier may either be an integer user-id, or an email address
            controlled by the user.  The type of identifier is auto-detected
            by looking for an `@` symbol, in which case it's taken as an email
            address, otherwise it's assumed to be an integer.
        :type user_identifier: string
        """
        user_manager = getUtility(IUserManager)
        if '@' in user_identifier:
            self._user = user_manager.get_user(user_identifier)
        else:
            # The identifier is the string representation of an integer that
            # must be converted to a UUID.
            try:
                user_id = UUID(int=int(user_identifier))
            except ValueError:
                self._user = None
            else:
                self._user = user_manager.get_user_by_id(user_id)

    def on_get(self, request, response):
        """Return a single user end-point."""
        if self._user is None:
            not_found(response)
        else:
            okay(response, self._resource_as_json(self._user))

    @child()
    def addresses(self, request, segments):
        """/users/<uid>/addresses"""
        if self._user is None:
            return NotFound(), []
        return UserAddresses(self._user)

    def on_delete(self, request, response):
        """Delete the named user, all her memberships, and addresses."""
        if self._user is None:
            not_found(response)
            return
        for member in self._user.memberships.members:
            member.unsubscribe()
        user_manager = getUtility(IUserManager)
        for address in self._user.addresses:
            user_manager.delete_address(address)
        user_manager.delete_user(self._user)
        no_content(response)

    @child()
    def preferences(self, request, segments):
        """/addresses/<email>/preferences"""
        if len(segments) != 0:
            return BadRequest(), []
        if self._user is None:
            return NotFound(), []
        child = Preferences(
            self._user.preferences,
            'users/{0}'.format(self._user.user_id.int))
        return child, []

    def on_patch(self, request, response):
        """Patch the user's configuration (i.e. partial update)."""
        if self._user is None:
            not_found(response)
            return
        try:
            validator = PatchValidator(request, ATTRIBUTES)
        except UnknownPATCHRequestError as error:
            bad_request(
                response, b'Unknown attribute: {0}'.format(error.attribute))
        except ReadOnlyPATCHRequestError as error:
            bad_request(
                response, b'Read-only attribute: {0}'.format(error.attribute))
        else:
            validator.update(self._user, request)
            no_content(response)

    def on_put(self, request, response):
        """Put the user's configuration (i.e. full update)."""
        if self._user is None:
            not_found(response)
            return
        validator = Validator(**ATTRIBUTES)
        try:
            validator.update(self._user, request)
        except UnknownPATCHRequestError as error:
            bad_request(
                response, b'Unknown attribute: {0}'.format(error.attribute))
        except ReadOnlyPATCHRequestError as error:
            bad_request(
                response, b'Read-only attribute: {0}'.format(error.attribute))
        except ValueError as error:
            bad_request(response, str(error))
        else:
            no_content(response)

    @child()
    def login(self, request, segments):
        """Log the user in, sort of, by verifying a given password."""
        if self._user is None:
            return NotFound(), []
        return Login(self._user)



class AddressUser(_UserBase):
    """The user linked to an address."""

    def __init__(self, address):
        self._address = address
        self._user = address.user

    def on_get(self, request, response):
        """Return a single user end-point."""
        if self._user is None:
            not_found(response)
        else:
            okay(response, self._resource_as_json(self._user))

    def on_delete(self, request, response):
        """Delete the named user, all her memberships, and addresses."""
        if self._user is None:
            not_found(response)
            return
        self._user.unlink(self._address)
        no_content(response)

    def on_post(self, request, response):
        """Link a user to the address, and create it if needed."""
        if self._user:
            conflict(response)
            return
        # When creating a linked user by POSTing, the user either must already
        # exist, or it can be automatically created, if the auto_create flag
        # is given and true (if missing, it defaults to true).  However, in
        # this case we do not accept 'email' as a POST field.
        fields = CREATION_FIELDS.copy()
        del fields['email']
        fields['user_id'] = int
        fields['auto_create'] = as_boolean
        fields['_optional'] = fields['_optional'] + ('user_id', 'auto_create')
        try:
            validator = Validator(**fields)
            arguments = validator(request)
        except ValueError as error:
            bad_request(response, str(error))
            return
        user_manager = getUtility(IUserManager)
        if 'user_id' in arguments:
            raw_uid = arguments['user_id']
            user_id = UUID(int=raw_uid)
            user = user_manager.get_user_by_id(user_id)
            if user is None:
                not_found(response, b'No user with ID {}'.format(raw_uid))
                return
            okay(response)
        else:
            auto_create = arguments.pop('auto_create', True)
            if auto_create:
                # This sets the 201 or 400 status.
                user = create_user(arguments, response)
                if user is None:
                    return
            else:
                forbidden(response)
                return
        user.link(self._address)

    def on_put(self, request, response):
        """Set or replace the addresses's user."""
        if self._user:
            self._user.unlink(self._address)
        # Process post data and check for an existing user.
        fields = CREATION_FIELDS.copy()
        fields['user_id'] = int
        fields['_optional'] = fields['_optional'] + ('user_id', 'email')
        try:
            validator = Validator(**fields)
            arguments = validator(request)
        except ValueError as error:
            bad_request(response, str(error))
            return
        user_manager = getUtility(IUserManager)
        if 'user_id' in arguments:
            raw_uid = arguments['user_id']
            user_id = UUID(int=raw_uid)
            user = user_manager.get_user_by_id(user_id)
            if user is None:
                not_found(response, b'No user with ID {}'.format(raw_uid))
                return
            okay(response)
        else:
            user = create_user(arguments, response)
            if user is None:
                return
        user.link(self._address)



class Login:
    """<api>/users/<uid>/login"""

    def __init__(self, user):
        assert user is not None
        self._user = user

    def on_post(self, request, response):
        # We do not want to encrypt the plaintext password given in the POST
        # data.  That would hash the password, but we need to have the
        # plaintext in order to pass into passlib.
        validator = Validator(cleartext_password=GetterSetter(str))
        try:
            values = validator(request)
        except ValueError as error:
            bad_request(response, str(error))
            return
        is_valid, new_hash = config.password_context.verify(
            values['cleartext_password'], self._user.password)
        if is_valid:
            if new_hash is not None:
                self._user.password = new_hash
            no_content(response)
        else:
            forbidden(response)