Add TOTP support

This patch adds support for MFA TOTP on openstack dashboard.
A new configuration variable OPENSTACK_KEYSTONE_MFA_TOTP_ENABLED
was added false by default.
If enabled, users needing TOTP are prompted with a new form.
keystone doc: https://docs.openstack.org/keystone/latest/admin/auth-totp.html
Demonstration video : https://youtu.be/prDJJdFoMpM

Change-Id: I1047102a379c8a900a5e6840096bb671da4fd2ff
Blueprint: #totp-support
Closes-Bug: #2030477
This commit is contained in:
Benjamin Lasseye 2023-06-08 07:00:33 +00:00
parent 3c5006efb4
commit cb74c8c08f
17 changed files with 457 additions and 2 deletions

View File

@ -607,6 +607,16 @@ endpoint when looking it up in the service catalog. This overrides
the ``OPENSTACK_ENDPOINT_TYPE`` parameter. If set to ``None``,
``OPENSTACK_ENDPOINT_TYPE`` is used for the identity endpoint.
OPENSTACK_KEYSTONE_MFA_TOTP_ENABLED
-----------------------------------
.. versionadded:: 23.2.1(Bobcat)
Default: ``False``
A boolean to activate TOTP support. If activated, the plugin must
be present in ``AUTHENTICATION_PLUGINS``.
OPENSTACK_HOST
--------------
@ -1276,6 +1286,18 @@ Default:
A list of authentication plugins to be used. In most cases, there is no need to
configure this.
If ``OPENSTACK_KEYSTONE_MFA_TOTP_ENABLED`` is true, then this should look
like this:
.. code-block:: python
[
'openstack_auth.plugin.totp.TotpPlugin',
'openstack_auth.plugin.password.PasswordPlugin',
'openstack_auth.plugin.token.TokenPlugin'
]
AUTHENTICATION_URLS
~~~~~~~~~~~~~~~~~~~

View File

@ -0,0 +1,59 @@
{% load i18n %}
{% block pre_login %}
<form id="" class="ng-pristine ng-valid ng-scope"
method="POST"
action=""
autocomplete="off"
ng-controller="hzLoginController">
{% csrf_token %}
{% endblock %}
<div class="panel panel-default">
<div class="panel-heading">
{% block login_header %}
<h3 class="login-title">
{% trans 'TOTP authentification' %}
</h3>
{% endblock %}
</div>
<div class="panel-body">
{% block login_body %}
{% comment %}
These fake fields are required to prevent Chrome v34+ from autofilling form.
{% endcomment %}
{% if HORIZON_CONFIG.password_autocomplete != "on" %}
<div class="fake_credentials" style="display: none">
<input type="passcode" name="fake_passcode" value="" />
</div>
{%endif%}
<fieldset hz-login-finder>
{% if logout_reason %}
{% if logout_status == "success" %}
<div class="form-group clearfix error help-block alert alert-success" id="logout_reason">
{% else %}
<div class="form-group clearfix error help-block alert alert-danger" id="logout_reason">
{% endif %}
<p>{{ logout_reason }}</p>
</div>
{% endif %}
{% include "horizon/common/_form_fields.html" %}
</fieldset>
{% endblock %}
</div>
<div class="panel-footer">
{% block login_footer %}
<button id="LoginBtn" type="submit" class="btn btn-primary pull-right">
<span>{% trans "Log in" %}</span>
</button>
<div class="clearfix"></div>
{% endblock %}
</div>
</div>
{% block post_login%}
</form>
{% endblock %}

View File

@ -0,0 +1,26 @@
{% extends 'auth/_totp_form.html' %}
{% load i18n %}
{% block pre_login %}
<div class="container login">
<div class="row">
<div class="col-xs-11 col-sm-8 col-md-6 col-lg-5 horizontal-center">
{{ block.super }}
{% endblock %}
{% block login_header %}
{% include 'auth/_splash.html' %}
{{ block.super }}
{% endblock %}
{% block login_footer %}
{{ block.super }}
{% include '_login_form_footer.html' %}
{% endblock %}
{% block post_login %}
{{ block.super }}
</div>
</div>
</div>
{% endblock %}

View File

@ -0,0 +1,14 @@
{% extends "base.html" %}
{% load i18n %}
{% block title %}{% trans "Log in" %}{% endblock %}
{% block body_id %}splash{% endblock %}
{% block content %}
{% include 'auth/_totp_page.html' %}
{% endblock %}
{% block footer %}
{% include '_login_footer.html' %}
{% endblock %}

View File

@ -178,3 +178,5 @@ POLICY_FILES_PATH = ''
POLICY_FILES = {}
POLICY_DIRS = {}
DEFAULT_POLICY_FILES = {}
OPENSTACK_KEYSTONE_MFA_TOTP_ENABLED = False

View File

@ -46,3 +46,7 @@ class KeystoneCredentialsException(KeystoneAuthException):
class KeystonePassExpiredException(KeystoneAuthException):
"""The password is expired and needs to be changed."""
class KeystoneTOTPRequired(KeystoneAuthException):
"""The passcode TOTP is requirede to authentificate."""

View File

@ -155,6 +155,15 @@ class Login(django_auth_forms.AuthenticationForm):
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.',
@ -244,3 +253,54 @@ class Password(forms.Form):
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

View File

@ -14,9 +14,11 @@ from openstack_auth.plugin.base import BasePlugin
from openstack_auth.plugin.k2k import K2KAuthPlugin
from openstack_auth.plugin.password import PasswordPlugin
from openstack_auth.plugin.token import TokenPlugin
from openstack_auth.plugin.totp import TotpPlugin
__all__ = ['BasePlugin',
'PasswordPlugin',
'TokenPlugin',
'K2KAuthPlugin']
'K2KAuthPlugin',
'TotpPlugin']

View File

@ -133,6 +133,24 @@ class BasePlugin(object, metaclass=abc.ABCMeta):
raise exc
msg = _('Invalid credentials.')
raise exceptions.KeystoneCredentialsException(msg)
except (keystone_exceptions.MissingAuthMethods) as exc:
msg = str(exc)
LOG.debug(msg)
try:
for required in exc.required_auth_methods:
if (len(required) == 2 and
'totp' in required and
'password' in required):
receipt = exc.receipt
msg = _('Authentication via TOTP is required.')
exc = exceptions.KeystoneTOTPRequired(msg)
exc.receipt = receipt
raise exc
msg = _("An error occurred authenticating. "
"Please try again later.")
raise exceptions.KeystoneAuthException(msg)
except Exception as exc:
raise exc
except (keystone_exceptions.ClientException,
keystone_exceptions.AuthorizationFailure) as exc:
msg = _("An error occurred authenticating. "

View File

@ -0,0 +1,43 @@
# 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 logging
from keystoneauth1.identity import v3 as v3_auth
from openstack_auth.plugin import base
LOG = logging.getLogger(__name__)
__all__ = ['TotpPlugin']
class TotpPlugin(base.BasePlugin):
def get_plugin(self, auth_url=None, username=None, passcode=None,
user_domain_name=None, receipt=None, **kwargs):
if not all((auth_url, username, passcode, user_domain_name, receipt)):
return None
LOG.debug('Attempting to authenticate with time-based one time'
' password for %s', username)
auth = v3_auth.TOTP(
auth_url=auth_url,
username=username,
passcode=passcode,
user_domain_name=user_domain_name,
unscoped=True
)
auth.add_method(v3_auth.ReceiptMethod(receipt=receipt))
return auth

View File

@ -279,6 +279,48 @@ def generate_test_data(service_providers=False, endpoint='localhost'):
body=unscoped_token_dict
)
# TOTP
unscoped_totp_token_dict = {
'token': {
'methods': ['password', 'totp'],
'expires_at': expiration,
'user': {
'id': user_dict['id'],
'name': user_dict['name'],
'domain': {
'id': domain_dict['id'],
'name': domain_dict['name']
},
},
'catalog': [keystone_service]
}
}
if service_providers:
unscoped_totp_token_dict['token']['service_providers'] = sp_list
test_data.unscoped_access_info_totp = access.create(
resp=auth_response,
body=unscoped_totp_token_dict
)
missing_methods_response_headers = {
'X-Subject-Token': auth_token,
'Openstack-Auth-Receipt': auth_token
}
missing_methods_response_text = """{
"required_auth_methods": [["totp", "password"]],
"receipt": {
"methods": ["password"],
"expires_at": "2023-08-15T10:31:53.000000Z"
}
}"""
test_data.missing_methods_response = TestResponse({
"headers": missing_methods_response_headers,
"status_code": 401,
"text": missing_methods_response_text
})
# Service Catalog
test_data.service_catalog = service_catalog.ServiceCatalogV3(
[keystone_service, nova_service])

View File

@ -0,0 +1,11 @@
<!DOCTYPE html>
<html>
<head>
<title>TOTP</title>
</head>
<body>
<form action="." method="POST">{{ csrf_token }}
{{ form.as_p }}
</form>
</body>
</html>

View File

@ -1441,3 +1441,98 @@ class OpenStackAuthTestsInternalURL(OpenStackAuthTests):
class OpenStackAuthTestsAdminURL(OpenStackAuthTests):
interface = 'adminURL'
class OpenstackAuthTestsTOTP(test.TestCase):
interface = None
def setUp(self):
super().setUp()
params = {
'OPENSTACK_API_VERSIONS': {'identity': 3},
'OPENSTACK_KEYSTONE_URL': "http://localhost/identity/v3",
'OPENSTACK_KEYSTONE_MFA_TOTP_ENABLED': True,
'AUTHENTICATION_PLUGINS': [
'openstack_auth.plugin.totp.TotpPlugin',
'openstack_auth.plugin.password.PasswordPlugin',
'openstack_auth.plugin.token.TokenPlugin'],
}
if self.interface:
params['OPENSTACK_ENDPOINT_TYPE'] = self.interface
override = self.settings(**params)
override.enable()
self.addCleanup(override.disable)
self.data = data_v3.generate_test_data()
def get_form_data(self, user):
return {'region': "default",
'domain': DEFAULT_DOMAIN,
'password': user.password,
'username': user.name}
def get_form_totp_data(self):
return {'totp': "000000"}
def test_totp_form(self):
user = self.data.user
url = reverse('totp', args=[user.name])
response = self.client.get(url)
self.assertEqual(response.status_code, 200)
self.assertContains(response, 'totp')
@mock.patch('keystoneauth1.identity.v3.Password.get_access')
def test_totp_redirect(self, mock_get_access):
user = self.data.user
response = self.data.missing_methods_response
form_data = self.get_form_data(user)
url = reverse('login')
mock_get_access.side_effect = keystone_exceptions.MissingAuthMethods(
response)
# GET the page to set the test cookie.
response = self.client.get(url, form_data)
self.assertEqual(response.status_code, 200)
# POST to the page to acces at TOTP authentification page.
response = self.client.post(url, form_data)
self.assertEqual(response.status_code, 302)
self.assertEqual(response.url, "/totp/%s/" % user.name)
@mock.patch('keystoneauth1.identity.v3.Token.get_access')
@mock.patch('keystoneauth1.identity.v3.BaseAuth.get_access')
@mock.patch('keystoneauth1.identity.v3.Password.get_access')
@mock.patch('keystoneclient.v3.client.Client')
def test_totp_login(self, mock_client, mock_get_access, mock_get_access_,
mock_get_access_token):
projects = [self.data.project_one, self.data.project_two]
user = self.data.user
response = self.data.missing_methods_response
form_data = self.get_form_data(user)
url = reverse('login')
mock_get_access.side_effect = keystone_exceptions.MissingAuthMethods(
response)
# Set test cookie:
response = self.client.get(url, form_data)
response = self.client.post(url, form_data)
form_data = self.get_form_totp_data()
url = response.url
mock_get_access_.return_value = self.data.unscoped_access_info_totp
mock_client.return_value.projects.list.return_value = projects
mock_get_access_token.return_value = self.data.unscoped_access_info_totp
# GET the page to set the test cookie.
response = self.client.get(url, form_data)
self.assertEqual(response.status_code, 200)
# POST to the page to log in.
response = self.client.post(url, form_data)
self.assertRedirects(response, settings.LOGIN_REDIRECT_URL)

View File

@ -22,5 +22,9 @@ urlpatterns = [
re_path(r"", include('openstack_auth.urls')),
re_path(r"^websso/$", views.websso, name='websso'),
re_path(r"^$",
generic.TemplateView.as_view(template_name="auth/blank.html"))
generic.TemplateView.as_view(template_name="auth/blank.html")),
re_path(r'^totp/(?P<user_name>[^/]+)/$',
views.TotpView.as_view(),
name='totp')
]

View File

@ -42,6 +42,13 @@ if utils.allow_expired_passowrd_change():
name='password')
)
if settings.OPENSTACK_KEYSTONE_MFA_TOTP_ENABLED:
urlpatterns.append(
re_path(r'^totp/(?P<user_name>[^/]+)/$',
views.TotpView.as_view(),
name='totp')
)
if settings.WEBSSO_ENABLED:
urlpatterns += [
re_path(r"^websso/$", views.websso, name='websso'),

View File

@ -163,6 +163,11 @@ def login(request):
form_class=form,
extra_context=extra_context,
redirect_authenticated_user=False)(request)
except exceptions.KeystoneTOTPRequired as exc:
res = django_http.HttpResponseRedirect(
reverse('totp', args=[request.POST.get('username')]))
request.session['receipt'] = exc.receipt
request.session['domain'] = request.POST.get('domain')
except exceptions.KeystonePassExpiredException as exc:
res = django_http.HttpResponseRedirect(
reverse('password', args=[exc.user_id]))
@ -484,3 +489,40 @@ class PasswordView(edit_views.FormView):
res = django_http.HttpResponseRedirect(self.success_url)
set_logout_reason(res, msg)
return res
class TotpView(edit_views.FormView):
"""Logs a user in using a TOTP authentification"""
template_name = 'auth/totp.html'
form_class = forms.TimeBasedOneTimePassword
success_url = settings.LOGIN_REDIRECT_URL
fail_url = "/login/"
def get_initial(self):
return {
'request': self.request,
'username': self.kwargs['user_name'],
'receipt': self.request.session.get('receipt'),
'region': self.request.COOKIES.get('login_region'),
'domain': self.request.session.get('domain'),
}
def form_valid(self, form):
auth.login(self.request, form.user_cache)
res = django_http.HttpResponseRedirect(self.success_url)
request = self.request
if self.request.user.is_authenticated:
del request.session['receipt']
auth_user.set_session_from_user(request, request.user)
regions = dict(forms.get_region_choices())
region = request.user.endpoint
login_region = request.POST.get('region')
region_name = regions.get(login_region)
request.session['region_endpoint'] = region
request.session['region_name'] = region_name
return res
def form_invalid(self, form):
if 'KeystoneNoBackendException' in str(form.errors):
return django_http.HttpResponseRedirect(self.fail_url)
return super().form_invalid(form)

View File

@ -0,0 +1,4 @@
---
features:
- |
Add suport for Time-based One-time Passwords.