~roadmr/canonical-identity-provider/update-raven

# Copyright 2010 Canonical Ltd.  This software is licensed under the
# GNU Affero General Public License version 3 (see the file LICENSE).

import logging
import re

from django import forms
from django.conf import settings
from django.forms import Form, fields, widgets
from django.urls import reverse
from django.utils import translation
from django.utils.translation import ugettext_lazy as _
from gargoyle import gargoyle

from identityprovider.const import (
    AX_DATA_FIELDS,
    AX_DATA_LABELS,
)
from identityprovider.fields import (
    FIELD_REQUIRED,
    DisplayNameField,
    EmailField,
    OATHPasswordField,
    PasswordField,
    PreferredEmailField,
    UsernameField,
    default_errors,
)
from identityprovider.models import (
    Account,
    EmailAddress,
    get_team_memberships_for_user,
)
from identityprovider.models.const import AccountStatus, EmailStatus
from identityprovider.validators import get_password_validator
from webservices.launchpad import (
    NameAlreadyTaken,
    NotPlaceholderAccount,
    set_lp_username,
)


logger = logging.getLogger(__name__)

TWOFACTOR_CHOICES = (
    (True, _('Always require an authentication device')),
    (False, _('Require only when logging in to sites that request it')),
)

tos_error = {
    'required': _(
        u'Check the box to indicate that you agree with our terms of use:'
    ),
}

VALID_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9\+\.\-]+$")

INVALID_NAME = _(
    'Usernames must be at least two characters long and start with a letter '
    'or number. All letters must be lower-case. The characters +, - and . '
    'are also allowed after the first character.')


class DeleteAccountForm(forms.Form):
    pass


class GenericEmailForm(forms.Form):

    email = EmailField(placeholder=_('Your email address'))


class LoginForm(GenericEmailForm):

    password = PasswordField(placeholder=_('Password'))


class TwoFactorForm(forms.Form):
    """This form shows *only* the two factor oath token field."""
    oath_token = OATHPasswordField()

    def __init__(self, *args, **kwargs):
        user = kwargs.pop('user', None)
        super(TwoFactorForm, self).__init__(*args, **kwargs)
        self.sync_2fa_available = user and user.has_twofactor_hotp_devices()


class TwoFactorSyncForm(forms.Form):
    """Used to synchronize 2fa OTPs."""

    FIELD_QUANTITY = 3

    def __init__(self, *args, **kwargs):
        super(TwoFactorSyncForm, self).__init__(*args, **kwargs)
        for i in range(1, self.FIELD_QUANTITY + 1):
            field_name = 'oath_token{}'.format(i)
            self.fields[field_name] = OATHPasswordField(required=True)


class TwoFactorLoginForm(LoginForm, TwoFactorForm):
    """A form that requires both password and 2factor oath token for login."""


class ResetPasswordFormMixin(object):

    def __init__(self, *args, **kwargs):
        self.account = kwargs.pop('account', None)
        super(ResetPasswordFormMixin, self).__init__(*args, **kwargs)

        validate_policy = get_password_validator(self.account)
        self.fields['password'] = PasswordField(
            placeholder=validate_policy.PASSWORD_POLICY_HELP_TEXT,
            help_text=validate_policy.PASSWORD_POLICY_HELP_TEXT,
            validators=[validate_policy],
        )
        self.fields['passwordconfirm'] = PasswordField(
            placeholder=_('Retype password'))

    def clean_password(self):
        password = self.cleaned_data.get('password')
        if (not password and self.data.get('passwordconfirm')):
            raise forms.ValidationError(FIELD_REQUIRED)
        return password

    def clean_passwordconfirm(self):
        password = self.cleaned_data.get('passwordconfirm')
        if (not password and self.data.get('passwordconfirm')):
            raise forms.ValidationError(FIELD_REQUIRED)
        return password

    def clean(self):
        cleaned_data = super(ResetPasswordFormMixin, self).clean()
        password = cleaned_data.get('password')
        passwordconfirm = cleaned_data.get('passwordconfirm')
        if ((password or passwordconfirm) and
                password != passwordconfirm and
                not self['password'].errors and
                not self['passwordconfirm'].errors):
            raise forms.ValidationError(_("Passwords didn't match"))
        return cleaned_data


class ResetPasswordForm(ResetPasswordFormMixin, forms.Form):
    pass


class NewAccountForm(GenericEmailForm, ResetPasswordForm):

    displayname = DisplayNameField()
    accept_tos = fields.BooleanField(
        error_messages=tos_error, required=True,
        widget=widgets.CheckboxInput(attrs={'required': 'required'}),
    )

    def __init__(self, *args, **kwargs):
        self.require_username = kwargs.pop('require_username', False)
        self.require_password = kwargs.pop('require_password', True)
        super(NewAccountForm, self).__init__(*args, **kwargs)
        # validate password against email
        leak_creds_validator = get_password_validator(
            email=self.data.get('email', None))
        self.fields['password'].validators = [leak_creds_validator]
        self.fields['password'].required = self.require_password
        self.fields['passwordconfirm'].required = self.require_password

        enable_username = gargoyle.is_active('USERNAME_UI')

        if enable_username:
            placeholder = _('Username')
            self.fields['username'] = UsernameField(
                required=self.require_username, placeholder=placeholder)

    def clean_username(self):
        if 'username' not in self.cleaned_data:
            return None
        username = self.cleaned_data['username'].strip()
        if not username:
            return username
        if not VALID_NAME_RE.match(username):
            raise forms.ValidationError(INVALID_NAME)

        # Check with LP if the `username` can be used to continue with the
        # registration. We don't have the openid_identifier for this
        # registration yet since the account hasn't been created. Using a
        # dummy openid_identifier for this purpose seems appropriate here.
        # `setUsernameFromSSO` in LP does also check for the validity of the
        # `username` and raises `InvalidName` but that will not happen here
        # because of the above form validation.
        dummy_openid = "dummyID~"
        try:
            set_lp_username(dummy_openid, username, dry_run=True)
        except NameAlreadyTaken:
            raise forms.ValidationError(_('%s is already in use.') % username)
        except Exception as e:
            raise Exception(_(
                'Dry run of setting username(%s) with dummy openid_identifier'
                '(%s) in Launchpad failed with (%s).') % (
                    username, dummy_openid,
                    e.msg if hasattr(e, 'msg') else str(e)))

        return username

    def clean_password(self):
        # Unify `password=None` and `password=''` cases.
        if (not self.require_password and
                not self.cleaned_data.get('password')):
            return None
        return super(NewAccountForm, self).clean_password()


class EditAccountForm(ResetPasswordFormMixin, forms.ModelForm):

    displayname = DisplayNameField()
    twofactor_required = fields.BooleanField(
        required=False,
        widget=widgets.RadioSelect(choices=TWOFACTOR_CHOICES),
    )
    warn_about_backup_device = fields.BooleanField(
        label=_('Warn me about not having a backup device'),
        required=False)

    def __init__(self, instance, **kwargs):
        self.password_changed = False
        enable_device_prefs = kwargs.pop('enable_device_prefs', False)
        enable_username = kwargs.pop('enable_username', False)
        require_username = kwargs.pop('require_username', False)
        kwargs['account'] = instance
        super(EditAccountForm, self).__init__(instance=instance, **kwargs)
        assert self.account == self.instance

        self.fields['password'].required = False
        self.fields['passwordconfirm'].required = False

        if self.instance.preferredemail is None:
            preferredemail_id = None
        else:
            preferredemail_id = self.instance.preferredemail.id

        # check for emails that can be preferred
        validated_emails = self.instance.verified_emails()

        if validated_emails.count() > 0:
            emails_url = reverse('account-emails')
            text = _(
                'Only verified email addresses are listed. Click '
                '<a href="%s">Manage email addresses</a> '
                'to add and verify email addresses.') % emails_url

            # add and display a dropdown with the valid choices
            self.fields['preferred_email'] = PreferredEmailField(
                queryset=validated_emails.order_by('email'),
                initial=preferredemail_id,
                help_text=text,
            )

        if enable_username:
            disabled = not self.instance.is_verified
            if not disabled:
                placeholder = _('Username')
            else:
                placeholder = _('Verify your email address first')
            self.fields['username'] = UsernameField(
                required=False, placeholder=placeholder,
                disabled=disabled)
            if self.instance.person is not None or require_username:
                self.fields['username'].required = True

        # store for later (clean_* methods) the ability of this account to edit
        # or not 2fa settings -- for example, we enforce values for some
        # specific team memberships
        self.must_require_twofactor = False

        if not enable_device_prefs:
            self.fields.pop('twofactor_required')
            self.fields.pop('warn_about_backup_device')

        # call must_require_twofactor(), which queries about team membership,
        # only once per form instance
        elif self.instance.must_require_twofactor():
            self.must_require_twofactor = True
            for field in ('twofactor_required', 'warn_about_backup_device'):
                self.fields[field].widget.attrs['readonly'] = True
                self.fields[field].widget.attrs['disabled'] = True

        # Custom initial field error message must be set at the end of
        # __init__() as it depends on self._errors being initialized.
        if require_username and not self.is_bound:
            self.errors.setdefault('username', ['Field required'])

    class Meta:
        fields = ('displayname', 'warn_about_backup_device',
                  'twofactor_required')
        model = Account

    def clean_displayname(self):
        name = self.cleaned_data['displayname'].strip()
        if len(name) == 0:
            raise forms.ValidationError(FIELD_REQUIRED)
        return name

    def clean_username(self):
        if 'username' not in self.cleaned_data:
            return
        if not self.instance.is_verified:
            raise forms.ValidationError(
                _("Verify your email address first."))
        username = self.cleaned_data['username'].strip()
        if not username:
            return username
        if not VALID_NAME_RE.match(username):
            raise forms.ValidationError(INVALID_NAME)
        if username and username != self.instance.person_name:
            # The username is going to change, so ask LP to do a dry
            # run to see if it will fail.
            error = None
            try:
                set_lp_username(
                    self.instance.openid_identifier, username,
                    dry_run=True)
            except NameAlreadyTaken:
                error = _('%s is already in use.') % username
            except NotPlaceholderAccount:
                error = _(
                    'Because you have a Launchpad account, you need to '
                    'change your username from Launchpad.')
            if error is not None:
                raise forms.ValidationError(error)
        return username

    def clean_preferred_email(self):
        if 'preferred_email' in self.cleaned_data:
            email = self.cleaned_data['preferred_email']
            if email.status == EmailStatus.NEW:
                logger.debug("status is NEW.")
                raise forms.ValidationError(_("Please select an "
                                              "already validated address."))
            return email

    def clean_twofactor_required(self):
        if self.must_require_twofactor:
            self.cleaned_data['twofactor_required'] = True
        return self.cleaned_data['twofactor_required']

    def clean_warn_about_backup_device(self):
        if self.must_require_twofactor:
            self.cleaned_data['warn_about_backup_device'] = True
        return self.cleaned_data['warn_about_backup_device']

    def save(self):
        if 'username' in self.cleaned_data:
            new_username = self.cleaned_data['username']
            if new_username and new_username != self.instance.person_name:
                set_lp_username(
                    self.instance.openid_identifier, new_username)
                self.instance.need_username_refresh = True
        password = self.cleaned_data['password']
        if password:
            self.instance.set_password(password)
            self.password_changed = True
        if 'preferred_email' in self.cleaned_data:
            self.instance.preferredemail = self.cleaned_data['preferred_email']

        return super(EditAccountForm, self).save()


class NewEmailForm(forms.Form):

    newemail = EmailField(placeholder=_('Email address'))

    def __init__(self, *args, **kwargs):
        self.account = kwargs.pop('account', None)
        super(NewEmailForm, self).__init__(*args, **kwargs)

    def clean_newemail(self):
        data = self.cleaned_data['newemail']
        try:
            if self.account is not None:
                if (self.account.emailaddress_set.count() >=
                        settings.MAX_EMAILS_PER_ACCOUNT):
                    raise forms.ValidationError(_(
                        'Account reached the maximum allowed number of email '
                        'addresses (%d).') % settings.MAX_EMAILS_PER_ACCOUNT)

                # check email was not previously invalidated for this account
                invalidated = self.account.invalidatedemailaddress_set.filter(
                    email__iexact=data)
                if invalidated.exists():
                    raise forms.ValidationError(_(
                        "Email previously invalidated for this account."
                    ))
            email = EmailAddress.objects.get(email__iexact=data)
            if email.status != EmailStatus.NEW:
                raise forms.ValidationError(_("Please verify your email."))
            else:
                raise forms.ValidationError(_("Email already registered."))
        except EmailAddress.DoesNotExist:
            return data
        return data


class PreAuthorizeForm(forms.Form):
    trust_root = forms.CharField(error_messages=default_errors)
    callback = forms.CharField(error_messages=default_errors)


class UserAttribsRequestForm(Form):
    """A form object for user control over OpenID Attribute Exchange."""

    def __init__(self, request, sreg_request, ax_request, rpconfig,
                 approved_data=None):
        self.request = request
        self.sreg_request = sreg_request
        self.ax_request = ax_request
        self.rpconfig = rpconfig
        self.approved_data = approved_data
        # generate initial form data
        self._split_and_filter_requested_attributes()
        self._get_data_for_user()
        super(UserAttribsRequestForm, self).__init__(self.data)
        # Use blank suffix, ala pre 1.6 behaviour, for template compatiblity
        self.label_suffix = ''
        self._init_fields(self.data)

    def _split_and_filter_requested_attributes(self):
        # Merge the requested attributes from sreg_request and ax_request and
        # filter out any that we don't recognise or that aren't allowed by our
        # rpconfig.
        # The rule is that if at least one request lists it as required, it's
        # required, otherwise it's optional.
        known_attribs = set([a for a in AX_DATA_FIELDS.iterAliases()])
        if self.rpconfig is not None:
            allowed = self.rpconfig.allowed_user_attribs or ''
            allowed_attribs = known_attribs.intersection(allowed.split(','))
        else:
            allowed_attribs = known_attribs

        required = set()
        optional = set()
        if self.sreg_request:
            required.update(self.sreg_request.required)
            optional.update(self.sreg_request.optional)
        if self.ax_request:
            for uri, attr in self.ax_request.requested_attributes.iteritems():
                if attr.required:
                    required.add(AX_DATA_FIELDS.getAlias(uri))
                else:
                    optional.add(AX_DATA_FIELDS.getAlias(uri))
        optional.difference_update(required)
        self.required = required.intersection(allowed_attribs)
        self.optional = optional.intersection(allowed_attribs)

    def _get_data_for_user(self):
        """Data to ask about in the form based on the user's account record."""
        values = {}
        user = self.request.user
        values['fullname'] = user.displayname
        if user.preferredemail is not None:
            values['email'] = user.preferredemail.email
        if user.person_name:
            values['nickname'] = user.person_name
        if user.person is not None:
            if user.person.time_zone is not None:
                values['timezone'] = user.person.time_zone
        if user.preferredlanguage is not None:
            values['language'] = user.preferredlanguage
        else:
            values['language'] = translation.get_language_from_request(
                self.request)
        values['account_verified'] = (
            'token_via_email' if user.is_verified else 'no')
        logger.debug('user attrib values = %s', str(values))

        self.data = dict([(f, values[f]) for f in self.required | self.optional
                          if f in values])

    def _init_fields(self, data):
        """Initialises form fields for the user's ax data.
        """
        for key, val in data.items():
            label = "%s: %s" % (AX_DATA_LABELS.get(key, key), val)
            attrs = {}
            if (key in self.required):
                attrs['class'] = 'required'
                if self.rpconfig is not None:
                    attrs['disabled'] = 'disabled'
            self.fields[key] = fields.BooleanField(
                label=label,
                widget=forms.CheckboxInput(attrs=attrs,
                                           check_test=self.check_test(key)),
            )

    def check_test(self, name):
        """Determines if a checkbox should be pre-checked based on previously
        approved user data, openid request and relying party type.
        """
        def inner(value):
            # Don't approve fields that weren't requested
            if name not in (self.required | self.optional):
                return False

            if self.rpconfig:
                # Trusted site, check required fields
                if (name in self.required):
                    return True
                if self.approved_data and name in self.approved_data.get(
                        'requested', []):
                    # The field was previously requested, use the same response
                    return name in self.approved_data.get('approved', [])
                # We've never (dis)approved this field before, default to True
                return True
            else:
                # If we have previous (dis)approval for this site, use it
                if self.approved_data and name in self.approved_data.get(
                        'requested', []):
                    return name in self.approved_data.get('approved', [])
                # No previous (dis)approval, check required and leave the rest
                return name in self.required
        return inner

    def field_approved(self, field):
        """Check if the field should be returned in the response based on user
        preferences and overridden for trusted relying parties.
        """
        post = self.request.POST
        approved = set([k for k in post.keys() if post[k]])
        if self.rpconfig is not None:
            approved.update(self.required)
            if self.rpconfig.auto_authorize:
                approved.update(self.optional)
        return field in approved

    @property
    def data_approved_for_request(self):
        """Return the list of user attributes approved for the request."""
        return dict(
            [(f, self.data[f]) for f in self.data if self.field_approved(f)])

    @property
    def has_data(self):
        """Helper property to check if this form has any data."""
        return len(self.data) > 0


class TeamsRequestForm(Form):
    """A form object for user control over OpenID teams data."""

    fields = {}

    @property
    def teams_approved_by_user(self):
        """Get the list of teams actually approved by the user."""
        if self.request_method == 'POST':
            return [t for t in self.memberships if t in self.request.POST]
        else:
            return []

    def __init__(self, request, teams_request, rpconfig, approved_data=None):
        self.request = request
        self.request_method = request.META.get('REQUEST_METHOD')
        self.teams_request = teams_request
        self.rpconfig = rpconfig
        self.approved_data = approved_data
        self.memberships = self._get_teams_for_user(
            request.user, rpconfig and rpconfig.can_query_any_team)

        super(TeamsRequestForm, self).__init__(self.memberships)

        # Use blank suffix, ala pre 1.6 behaviour, for template compatiblity
        self.label_suffix = ''
        self._init_fields(self.memberships)

    def _get_teams_for_user(self, user, include_private=False):
        """Get the list of teams to ask about in the form based on the user's
        team membership.
        """
        all_teams = self.teams_request.allRequestedTeams()
        memberships = get_team_memberships_for_user(all_teams, user,
                                                    include_private)
        return dict((t, t) for t in memberships)

    def _init_fields(self, form_data):
        """Initialises form fields for the user's team memberships.
        """
        if len(form_data) == 1:
            label_format = 'Team membership: %s'
        else:
            label_format = '%s'
        for team in form_data:
            label = label_format % team
            self.fields[team] = fields.BooleanField(
                label=label, widget=forms.CheckboxInput(
                    check_test=self.check_test(team)))

    def check_test(self, name):
        """Determines if a checkbox should be pre-checked based on previously
        approved user data and relying party type.
        """
        def inner(value):
            if (self.approved_data and
                    name in self.approved_data.get('requested', [])):
                return name in self.approved_data.get('approved', [])
            else:
                return self.rpconfig is not None
        return inner

    @property
    def has_data(self):
        """Helper property to check if this form has any data."""
        return len(self.memberships) > 0


class OATHDeviceForm(Form):
    name = fields.CharField()
    otp = OATHPasswordField()


class DeviceRenameForm(Form):
    name = fields.CharField()


class AccountStatusForm(Form):
    status = fields.ChoiceField(choices=[(attr, attr)
                                for attr in AccountStatus.enum()])
    status_comment = fields.CharField()

    def clean_status(self):
        status = self.cleaned_data['status']
        return getattr(AccountStatus, status)


class MacaroonRequestForm(Form):
    """A form object for user control over requesting discharge macaroons."""

    def __init__(self, request, macaroon_request, rpconfig, trust_root,
                 approved_data=None):
        self.request = request
        self.macaroon_request = macaroon_request
        self.caveat_id = self.macaroon_request.caveat_id
        self.rpconfig = rpconfig
        self.trust_root = trust_root
        self.approved_data = approved_data
        self.data = self._get_data_for_user()
        super(MacaroonRequestForm, self).__init__(self.data)
        # Use blank suffix, a la pre 1.6 behaviour, for template compatibility
        self.label_suffix = ''
        self._init_fields(self.data)

    def _get_data_for_user(self):
        """Data to ask about in the form."""
        data = {}
        if self.caveat_id:
            data['macaroon'] = self.caveat_id
        return data

    def _init_fields(self, data):
        """Initialise fields from the discharge macaroon request."""
        for location, caveat_id in data.iteritems():
            if self.rpconfig:
                rp_displayname = self.rpconfig.displayname
            else:
                rp_displayname = self.trust_root
            label = 'Service authorization for %s' % rp_displayname
            self.fields[location] = fields.BooleanField(
                label=label, widget=forms.CheckboxInput(
                    check_test=self.check_test(location)))

    def check_test(self, name):
        """Determine if a checkbox should be pre-checked."""
        def inner(value):
            if name not in self.data:
                return False

            # If we have previous (dis)approval for this site, use it.
            if (self.approved_data and
                    name in self.approved_data.get('requested', [])):
                return name in self.approved_data.get('approved', [])
            # No previous (dis)approval; default to True.
            return True

        return inner

    def field_approved(self, field):
        """Return whether `field` can be included in a response.

        Check if the field should be returned in the response based on user
        preferences and overridden for trusted relying parties.
        """
        post = self.request.POST
        approved = set(k for k, v in post.iteritems() if v)
        if self.rpconfig is not None:
            approved.update(self.data)
        return field in approved

    @property
    def data_approved_for_request(self):
        """Return the list of user attributes approved for the request."""
        return {
            f: v for f, v in self.data.iteritems() if self.field_approved(f)}

    @property
    def has_data(self):
        """Helper property to check if this form has any data."""
        return len(self.data) > 0