diff options
Diffstat (limited to 'src/mailman/rest/users.py')
| -rw-r--r-- | src/mailman/rest/users.py | 140 |
1 files changed, 119 insertions, 21 deletions
diff --git a/src/mailman/rest/users.py b/src/mailman/rest/users.py index cfea36cfa..80c53b565 100644 --- a/src/mailman/rest/users.py +++ b/src/mailman/rest/users.py @@ -39,7 +39,8 @@ from mailman.interfaces.usermanager import IUserManager from mailman.rest.addresses import UserAddresses from mailman.rest.helpers import ( BadRequest, CollectionMixin, GetterSetter, NotFound, bad_request, child, - created, etag, forbidden, no_content, not_found, okay, paginate, path_to) + created, etag, forbidden, no_content, not_found, okay, paginate, path_to, + conflict) from mailman.rest.preferences import Preferences from mailman.rest.validator import PatchValidator, Validator @@ -62,6 +63,34 @@ ATTRIBUTES = dict( cleartext_password=PasswordEncrypterGetterSetter(), ) +CREATION_FIELDS = dict( + email=unicode, + display_name=unicode, + password=unicode, + _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, b'Address already exists: {0}'.format(error.address)) + return + 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/{0}'.format(user.user_id.int)) + created(response, location) + return user + class _UserBase(CollectionMixin): @@ -105,30 +134,12 @@ class AllUsers(_UserBase): def on_post(self, request, response): """Create a new user.""" try: - validator = Validator(email=unicode, - display_name=unicode, - password=unicode, - _optional=('display_name', 'password')) + validator = Validator(**CREATION_FIELDS) arguments = validator(request) except ValueError as error: bad_request(response, str(error)) return - # 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, b'Address already exists: {0}'.format(error.address)) - return - 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/{0}'.format(user.user_id.int)) - created(response, location) + create_user(arguments, response) @@ -242,6 +253,93 @@ class AUser(_UserBase): +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 + # Check for an existing user + fields = CREATION_FIELDS.copy() + del fields["email"] + fields["user_id"] = int + fields["autocreate"] = int + fields["_optional"] = fields["_optional"] + ('user_id', 'autocreate') + 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: + user_id = UUID(int=arguments["user_id"]) + user = user_manager.get_user_by_id(user_id) + if user is None: + not_found(response, b"No user with ID {0}".format( + arguments["user_id"])) + return + okay(response) + else: + if arguments.get("autocreate", True): # autocreate by default + arguments.pop("autocreate", None) + user = create_user(arguments, response) # will set "201 Created" + else: + not_found(response, b"No such user, and automatic " + "creation is disabled.") + return + user.link(self._address) + + def on_put(self, request, response): + """Set or replace the addresse'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: + user_id = UUID(int=arguments["user_id"]) + user = user_manager.get_user_by_id(user_id) + if user is None: + not_found(response, b"No user with ID {0}".format( + arguments["user_id"])) + return + okay(response) + else: + user = create_user(arguments, response) # will set "201 Created" + user.link(self._address) + + + class Login: """<api>/users/<uid>/login""" |