FreedomBox/plinth/modules/users/forms.py

#
# This file is part of FreedomBox.
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program 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 Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
#

import subprocess

from django import forms
from django.contrib import auth, messages
from django.contrib.auth.forms import SetPasswordForm, UserCreationForm
from django.contrib.auth.models import Group, User
from django.core.exceptions import ValidationError
from django.utils import translation
from django.utils.translation import ugettext as _
from django.utils.translation import ugettext_lazy

from plinth import actions, module_loader
from plinth import forms as plinthForms
from plinth.errors import ActionError
from plinth.modules import first_boot, users
from plinth.modules.security import set_restricted_access
from plinth.utils import is_user_admin
from plinth.models import UserProfile


def get_group_choices():
    """Return localized group description and group name in one string."""
    admin_group = ('admin', _('Access to all services and system settings'))
    users.register_group(admin_group)
    choices = {(g[0], ('{} ({})'.format(g[1], g[0]))) for g in users.groups}
    return sorted(list(choices), key=lambda g: g[0])


class ValidNewUsernameCheckMixin(object):
    """Mixin to check if a username is valid for created new user."""

    def clean_username(self):
        """Check for username collisions with system users."""
        username = self.cleaned_data['username']
        if self.instance.username != username and \
                not self.is_valid_new_username():
            raise ValidationError(
                _('Username is taken or is reserved.'), code='invalid')

        return username

    def is_valid_new_username(self):
        """Check for username collisions with system users."""
        username = self.cleaned_data['username']
        try:
            subprocess.run(['getent', 'passwd', username],
                           stdout=subprocess.DEVNULL, check=True)
            # Exit code 0 means that the username is already in use.
            return False
        except subprocess.CalledProcessError:
            pass

        for module_name, module in module_loader.loaded_modules.items():
            for reserved_username in getattr(module, 'reserved_usernames', []):
                if username == reserved_username:
                    return False

        return True


class CreateUserForm(ValidNewUsernameCheckMixin, UserCreationForm, plinthForms.LanguageSelectionForm):
    """Custom user create form.

    Include options to add user to groups.
    """
    groups = forms.MultipleChoiceField(
        choices=get_group_choices(), label=ugettext_lazy('Permissions'),
        required=False,
        widget=forms.CheckboxSelectMultiple, help_text=ugettext_lazy(
            'Select which services should be available to the new '
            'user. The user will be able to log in to services that '
            'support single sign-on through LDAP, if they are in the '
            'appropriate group.<br /><br />Users in the admin group '
            'will be able to log in to all services. They can also '
            'log in to the system through SSH and have '
            'administrative privileges (sudo).'))

    class Meta(UserCreationForm.Meta):
        fields = ('username', 'password1', 'password2', 'groups', 'language')

    def __init__(self, request, *args, **kwargs):
        """Initialize the form with extra request argument."""
        self.request = request
        super(CreateUserForm, self).__init__(*args, **kwargs)
        self.fields['groups'].choices = get_group_choices()

    def save(self, commit=True):
        """Save the user model and create LDAP user if required."""
        user = super(CreateUserForm, self).save(commit)
        UserProfile(user=user, preferred_language=self.cleaned_data['language']).save()

        if commit:
            try:
                actions.superuser_run('users', [
                    'create-user', user.get_username()
                ], input=self.cleaned_data['password1'].encode())
            except ActionError:
                messages.error(self.request, _('Creating LDAP user failed.'))

            for group in self.cleaned_data['groups']:
                try:
                    actions.superuser_run('users', [
                        'add-user-to-group',
                        user.get_username(), group
                    ])
                except ActionError:
                    messages.error(
                        self.request,
                        _('Failed to add new user to {group} group.').format(
                            group=group))

                group_object, created = Group.objects.get_or_create(name=group)
                group_object.user_set.add(user)

        return user


class UserUpdateForm(ValidNewUsernameCheckMixin, forms.ModelForm, plinthForms.LanguageSelectionForm):
    """When user info is changed, also updates LDAP user."""
    ssh_keys = forms.CharField(
        label=ugettext_lazy('SSH Keys'),
        required=False, widget=forms.Textarea, help_text=ugettext_lazy(
            'Setting an SSH public key will allow this user to '
            'securely log in to the system without using a '
            'password. You may enter multiple keys, one on each '
            'line. Blank lines and lines starting with # will be '
            'ignored.'))

    class Meta:
        """Metadata to control automatic form building."""
        fields = ('username', 'groups', 'ssh_keys', 'language', 'is_active')
        model = User
        widgets = {
            'groups': forms.widgets.CheckboxSelectMultiple(),
        }

    def __init__(self, request, username, *args, **kwargs):
        """Initialize the form with extra request argument."""
        group_choices = dict(get_group_choices())
        for group in group_choices:
            Group.objects.get_or_create(name=group)

        self.request = request
        self.username = username
        super(UserUpdateForm, self).__init__(*args, **kwargs)

        choices = []

        for c in sorted(self.fields['groups'].choices, key=lambda x: x[1]):
            # Handle case where groups exist in database for
            # applications not installed yet.
            if c[1] in group_choices:
                # Replace group names with descriptions
                choices.append((c[0], group_choices[c[1]]))

        self.fields['groups'].label = 'Permissions'
        self.fields['groups'].choices = choices

        if not is_user_admin(request):
            self.fields['is_active'].widget = forms.HiddenInput()
            self.fields['groups'].disabled = True

    def save(self, commit=True):
        """Update LDAP user name and groups after saving user model."""
        user = super(UserUpdateForm, self).save(commit=False)
        user.userprofile.preferred_language = self.cleaned_data['language']
        user.userprofile.save()
        user.save()

        # If user is updating their own profile then only translate the pages
        if self.username == self.request.user.username:
            self.request.session[translation.LANGUAGE_SESSION_KEY] = user.userprofile.preferred_language

        if commit:
            output = actions.superuser_run('users',
                                           ['get-user-groups', self.username])
            old_groups = output.strip().split('\n')
            old_groups = [group for group in old_groups if group]

            if self.username != user.get_username():
                try:
                    actions.superuser_run('users', [
                        'rename-user', self.username,
                        user.get_username()
                    ])
                except ActionError:
                    messages.error(self.request,
                                   _('Renaming LDAP user failed.'))

            new_groups = user.groups.values_list('name', flat=True)
            for old_group in old_groups:
                if old_group not in new_groups:
                    try:
                        actions.superuser_run('users', [
                            'remove-user-from-group',
                            user.get_username(), old_group
                        ])
                    except ActionError:
                        messages.error(self.request,
                                       _('Failed to remove user from group.'))

            for new_group in new_groups:
                if new_group not in old_groups:
                    try:
                        actions.superuser_run('users', [
                            'add-user-to-group',
                            user.get_username(), new_group
                        ])
                    except ActionError:
                        messages.error(self.request,
                                       _('Failed to add user to group.'))

            try:
                actions.superuser_run('ssh', [
                    'set-keys', '--username',
                    user.get_username(), '--keys',
                    self.cleaned_data['ssh_keys'].strip()
                ])
            except ActionError:
                messages.error(self.request, _('Unable to set SSH keys.'))

        return user


class UserChangePasswordForm(SetPasswordForm):
    """Custom form that also updates password for LDAP users."""

    def __init__(self, request, *args, **kwargs):
        """Initialize the form with extra request argument."""
        self.request = request
        super(UserChangePasswordForm, self).__init__(*args, **kwargs)

    def save(self, commit=True):
        """Save the user model and change LDAP password as well."""
        user = super(UserChangePasswordForm, self).save(commit)
        if commit:
            try:
                actions.superuser_run('users', [
                    'set-user-password',
                    user.get_username()
                ], input=self.cleaned_data['new_password1'].encode())
            except ActionError:
                messages.error(self.request,
                               _('Changing LDAP user password failed.'))

        return user


class FirstBootForm(ValidNewUsernameCheckMixin, auth.forms.UserCreationForm):
    """User module first boot step: create a new admin user."""

    def __init__(self, *args, **kwargs):
        self.request = kwargs.pop('request')
        super().__init__(*args, **kwargs)

    def save(self, commit=True):
        """Create and log the user in."""
        user = super().save(commit=commit)
        UserProfile(user=user).save()
        if commit:
            first_boot.mark_step_done('users_firstboot')

            try:
                actions.superuser_run('users', [
                    'create-user', user.get_username()
                ], input=self.cleaned_data['password1'].encode())
            except ActionError:
                messages.error(self.request, _('Creating LDAP user failed.'))

            try:
                actions.superuser_run('users', [
                    'add-user-to-group',
                    user.get_username(), 'admin'
                ])
            except ActionError:
                messages.error(self.request,
                               _('Failed to add new user to admin group.'))

            # Create initial Django groups
            for group_choice in get_group_choices():
                auth.models.Group.objects.get_or_create(name=group_choice[0])

            admin_group = auth.models.Group.objects.get(name='admin')
            admin_group.user_set.add(user)

            self.login_user(self.cleaned_data['username'],
                            self.cleaned_data['password1'])

            # Restrict console login to users in admin or sudo group
            try:
                set_restricted_access(True)
            except Exception:
                messages.error(self.request,
                               _('Failed to restrict console access.'))

        return user

    def login_user(self, username, password):
        """Try to login the user with the credentials provided"""
        try:
            user = auth.authenticate(username=username, password=password)
            auth.login(self.request, user)
        except Exception:
            pass
        else:
            message = _('User account created, you are now logged in')
            messages.success(self.request, message)