horizon/openstack_auth/forms.py

# 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.

import collections
import logging

from django.conf import settings
from django.contrib.auth import authenticate
from django.contrib.auth import forms as django_auth_forms
from django import forms
from django.utils.translation import gettext_lazy as _
from django.views.decorators.debug import sensitive_variables

from keystoneauth1 import plugin as auth_plugin
from openstack_auth import exceptions
from openstack_auth import utils


LOG = logging.getLogger(__name__)


def get_region_endpoint(region_id):
    if region_id == "default":
        return settings.OPENSTACK_KEYSTONE_URL
    all_regions = settings.AVAILABLE_REGIONS
    return all_regions[int(region_id)][0]


def get_region_choices():
    all_regions = settings.AVAILABLE_REGIONS
    if all_regions:
        regions = [("%d" % i, name) for i, (url, name) in
                   enumerate(all_regions)]
    else:
        regions = [("default", _("Default Region"))]
    return regions


class Login(django_auth_forms.AuthenticationForm):
    """Form used for logging in a user.

    Handles authentication with Keystone by providing the domain name, username
    and password. A scoped token is fetched after successful authentication.

    A domain name is required if authenticating with Keystone V3 running
    multi-domain configuration.

    If the user authenticated has a default project set, the token will be
    automatically scoped to their default project.

    If the user authenticated has no default project set, the authentication
    backend will try to scope to the projects returned from the user's assigned
    projects. The first successful project scoped will be returned.

    Inherits from the base ``django.contrib.auth.forms.AuthenticationForm``
    class for added security features.
    """
    use_required_attribute = False
    required_css_class = "required"
    warnings = []
    region = forms.ChoiceField(label=_("Region"), required=False)
    username = forms.CharField(
        label=_("User Name"),
        widget=forms.TextInput(attrs={"autofocus": "autofocus"}))
    password = forms.CharField(label=_("Password"),
                               strip=False,
                               widget=forms.PasswordInput(render_value=False))

    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        fields_ordering = ['username', 'password', 'region']
        if settings.OPENSTACK_KEYSTONE_MULTIDOMAIN_SUPPORT:
            last_domain = self.request.COOKIES.get('login_domain', None)
            if settings.OPENSTACK_KEYSTONE_DOMAIN_DROPDOWN:
                self.fields['domain'] = forms.ChoiceField(
                    label=_("Domain"),
                    initial=last_domain,
                    choices=settings.OPENSTACK_KEYSTONE_DOMAIN_CHOICES)
            else:
                self.fields['domain'] = forms.CharField(
                    initial=last_domain,
                    label=_("Domain"))
            fields_ordering = ['username', 'password', 'domain', 'region']
        self.fields['region'].choices = get_region_choices()
        if len(self.fields['region'].choices) == 1:
            self.fields['region'].initial = self.fields['region'].choices[0][0]
            self.fields['region'].widget = forms.widgets.HiddenInput()
        elif len(self.fields['region'].choices) > 1:
            self.fields['region'].initial = self.request.COOKIES.get(
                'login_region')

        # if websso is enabled and keystone version supported
        # prepend the websso_choices select input to the form
        if settings.WEBSSO_ENABLED:
            initial = settings.WEBSSO_INITIAL_CHOICE
            self.fields['auth_type'] = forms.ChoiceField(
                label=_("Authenticate using"),
                choices=settings.WEBSSO_CHOICES,
                required=False,
                initial=initial)
            # add auth_type to the top of the list
            fields_ordering.insert(0, 'auth_type')

        # websso is enabled, but keystone version is not supported
        elif settings.WEBSSO_ENABLED:
            msg = ("Websso is enabled but horizon is not configured to work " +
                   "with keystone version 3 or above.")
            LOG.warning(msg)
        self.fields = collections.OrderedDict(
            (key, self.fields[key]) for key in fields_ordering)

    @sensitive_variables()
    def clean(self):
        default_domain = settings.OPENSTACK_KEYSTONE_DEFAULT_DOMAIN
        username = self.cleaned_data.get('username')
        password = self.cleaned_data.get('password')
        domain = self.cleaned_data.get('domain', default_domain)
        region_id = self.cleaned_data.get('region')
        try:
            region = get_region_endpoint(region_id)
        except (ValueError, IndexError, TypeError):
            raise forms.ValidationError("Invalid region %r" % region_id)
        self.cleaned_data['region'] = region

        if not (username and password):
            # Don't authenticate, just let the other validators handle it.
            return self.cleaned_data

        try:
            self.user_cache = authenticate(request=self.request,
                                           username=username,
                                           password=password,
                                           user_domain_name=domain,
                                           auth_url=region)
            LOG.info('Login successful for user "%(username)s" using domain '
                     '"%(domain)s", remote address %(remote_ip)s.',
                     {'username': username, 'domain': domain,
                      'remote_ip': utils.get_client_ip(self.request)})
        except exceptions.KeystonePassExpiredException as exc:
            LOG.info('Login failed for user "%(username)s" using domain '
                     '"%(domain)s", remote address %(remote_ip)s: password'
                     ' expired.',
                     {'username': username, 'domain': domain,
                      'remote_ip': utils.get_client_ip(self.request)})
            if utils.allow_expired_passowrd_change():
                raise
            raise forms.ValidationError(exc)
        except exceptions.KeystoneTOTPRequired as exc:
            LOG.info('Login failed for user "%(username)s" using domain '
                     '"%(domain)s", remote address %(remote_ip)s: TOTP'
                     'required.',
                     {'username': username, 'domain': domain,
                      'remote_ip': utils.get_client_ip(self.request)})
            if settings.OPENSTACK_KEYSTONE_MFA_TOTP_ENABLED:
                raise
            raise forms.ValidationError(exc)
        except exceptions.KeystoneAuthException as exc:
            LOG.info('Login failed for user "%(username)s" using domain '
                     '"%(domain)s", remote address %(remote_ip)s.',
                     {'username': username, 'domain': domain,
                      'remote_ip': utils.get_client_ip(self.request)})
            raise forms.ValidationError(exc)
        return self.cleaned_data


class DummyAuth(auth_plugin.BaseAuthPlugin):
    """A dummy Auth object

    It is needed for _KeystoneAdapter to get the user_id from, but otherwise
    behaves as if it doesn't exist (is falsy).
    """
    def __init__(self, user_id):
        self.user_id = user_id

    def __bool__(self):
        return False

    def get_headers(self, session, **kwargs):
        return {}


class Password(forms.Form):
    """Form used for changing user's password without having to log in."""
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        config = getattr(settings, 'HORIZON_CONFIG', {})
        self.fields = collections.OrderedDict([
            (
                'region',
                forms.ChoiceField(label=_("Region"), required=False)
            ), (
                'original_password',
                forms.CharField(label=_("Original password"),
                                strip=False,
                                widget=forms.PasswordInput(render_value=False))
            ), (
                'password',
                forms.RegexField(
                    label=_("New password"),
                    strip=False,
                    widget=forms.PasswordInput(render_value=False),
                    regex=config.get(
                        'password_validator', {}).get('regex', '.*'),
                    error_messages={
                        'invalid':
                        config.get('password_validator',
                                   {}).get('help_text',
                                           _("Password is not accepted"))}),
            ), (
                'confirm_password',
                forms.CharField(label=_("Confirm password"),
                                strip=False,
                                widget=forms.PasswordInput(render_value=False))
            ),
        ])
        self.fields['region'].choices = get_region_choices()
        if len(self.fields['region'].choices) == 1:
            self.fields['region'].initial = self.fields['region'].choices[0][0]
            self.fields['region'].widget = forms.widgets.HiddenInput()

    @sensitive_variables('password', 'confirm_password', 'original_password')
    def clean(self):
        if not self.is_valid():
            return self.cleaned_data

        region_id = self.cleaned_data.get('region')
        try:
            region = get_region_endpoint(region_id)
        except (ValueError, IndexError, TypeError):
            raise forms.ValidationError("Invalid region %r" % region_id)
        self.cleaned_data['region'] = region

        password = self.cleaned_data.get('password')
        original_password = self.cleaned_data.get('original_password')
        confirm_password = self.cleaned_data.get('confirm_password')
        if password != confirm_password:
            raise forms.ValidationError(_('Passwords do not match.'))
        if password == original_password:
            raise forms.ValidationError(
                _('Old password and new password must be different.'))

        # Doing it here, to be able to raise ValidationError on failure.
        user_id = self.initial['user_id']
        session = utils.get_session(auth=DummyAuth(user_id))
        Client = utils.get_keystone_client().Client
        client = Client(session=session, user_id=user_id,
                        auth_url=region, endpoint=region)
        # This is needed so that keystoneclient doesn't try to authenticate.
        client.users.client.endpoint_override = region
        try:
            client.users.update_password(original_password, password)
        except Exception as e:
            LOG.error("Unable to update password due to exception: %s",
                      e)
            raise forms.ValidationError(
                _("Unable to update the user password."))
        return self.cleaned_data


class TimeBasedOneTimePassword(forms.Form):
    """Form used for TOTP authentification"""
    def __init__(self, *args, **kwargs):
        super().__init__(*args, **kwargs)
        self.fields = collections.OrderedDict([
            (
                'totp',
                forms.CharField(label=_("Passcode"),
                                required=True,
                                widget=forms.TextInput(
                                attrs={"autofocus": "autofocus"}))
            )
        ])

    @sensitive_variables('totp')
    def clean(self):
        default_domain = settings.OPENSTACK_KEYSTONE_DEFAULT_DOMAIN
        request = self.initial['request']
        domain = self.initial['domain']
        if domain == "" or domain is None:
            domain = default_domain
        username = self.initial['username']
        receipt = self.initial['receipt']
        region_id = self.initial['region']
        passcode = self.cleaned_data.get('totp')
        try:
            region = get_region_endpoint(region_id)
            LOG.info(region)
        except (ValueError, IndexError, TypeError):
            raise forms.ValidationError("Invalid region %r" % region_id)
        try:
            self.cleaned_data['region'] = region
            self.user_cache = authenticate(request=request,
                                           receipt=receipt,
                                           username=username,
                                           passcode=passcode,
                                           user_domain_name=domain,
                                           auth_url=region)
            LOG.info('Login successful for user "%(username)s" using domain '
                     '"%(domain)s", remote address %(remote_ip)s.',
                     {'username': username, 'domain': domain,
                      'remote_ip': utils.get_client_ip(request)})
        except exceptions.KeystoneNoBackendException as exc:
            LOG.info(exc)
            raise forms.ValidationError('KeystoneNoBackendException')
        except Exception as exc:
            LOG.info(exc)
            raise forms.ValidationError(exc)
        return self.cleaned_data