Time-based One-time Password

Support TOTP as a distinct authentication mechanism from Password.

bp totp-auth

Co-Authored-By: David Stanek <dstanek@dstanek.com>
Change-Id: Ic0ccf89b9f35d3167a413b10f43be43cf892aead

doc/source/auth-totp.rst

@@ -0,0 +1,141 @@
+..
+      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.
+
+===================================
+Time-based One-time Password (TOTP)
+===================================
+
+----------------
+Configuring TOTP
+----------------
+
+TOTP is not enabled in Keystone by default.  To enable it add the ``totp``
+authentication method to the ``[auth]`` section in ``keystone.conf``:
+
+.. code-block:: ini
+
+    [auth]
+    methods = external,password,token,oauth1,totp
+
+For a user to have access to TOTP, he must have configured TOTP credentials in
+Keystone and a TOTP device (i.e. `Google Authenticator`_).
+
+.. _Google Authenticator: http://www.google.com/2step
+
+TOTP uses a base32 encoded string for the secret. The secret must be at least
+148 bits (16 bytes). The following python code can be used to generate a TOTP
+secret:
+
+.. code-block:: python
+
+    import base64
+    message = '1234567890123456'
+    print base64.b32encode(message).rstrip('=')
+
+Example output::
+
+    GEZDGNBVGY3TQOJQGEZDGNBVGY
+
+This generated secret can then be used to add new 'totp' credentials to a
+specific user.
+
+Create a TOTP credential
+------------------------
+
+Create ``totp`` credentials for user:
+
+.. code-block:: bash
+
+    USER_ID=b7793000f8d84c79af4e215e9da78654
+    SECRET=GEZDGNBVGY3TQOJQGEZDGNBVGY
+
+    curl -i \
+      -H "Content-Type: application/json" \
+      -d '
+    {
+        "credential": {
+            "blob": "'$SECRET'",
+            "type": "totp",
+            "user_id": "'$USER_ID'"
+        }
+    }' \
+      http://localhost:5000/v3/credentials ; echo
+
+Google Authenticator
+--------------------
+
+On a device install Google Authenticator and inside the app click on 'Set up
+account' and then click on 'Enter provided key'.  In the input fields enter
+account name and secret.  Optionally a QR code can be generated programatically
+to avoid having to type the information.
+
+QR code
+-------
+
+Create TOTP QR code for device:
+
+.. code-block:: python
+
+    import qrcode
+
+    secret='GEZDGNBVGY3TQOJQGEZDGNBVGY'
+    uri = 'otpauth://totp/{name}?secret={secret}&issuer={issuer}'.format(
+        name='name',
+        secret=secret,
+        issuer='Keystone')
+
+    img = qrcode.make(uri)
+    img.save('totp.png')
+
+In Google Authenticator app click on 'Set up account' and then click on 'Scan
+a barcode', and then scan the 'totp.png' image.  This should create a new TOTP
+entry in the application.
+
+----------------------
+Authenticate with TOTP
+----------------------
+
+Google Authenticator will generate a 6 digit PIN (passcode) every few seconds.
+Use the passcode and your user ID to authenticate using the ``totp`` method.
+
+Tokens
+======
+
+Default scope
+-------------
+
+Get a token with default scope (may be unscoped) using totp:
+
+.. code-block:: bash
+
+    USER_ID=b7793000f8d84c79af4e215e9da78654
+    PASSCODE=012345
+
+    curl -i \
+      -H "Content-Type: application/json" \
+      -d '
+    { "auth": {
+            "identity": {
+                "methods": [
+                    "totp"
+                ],
+                "totp": {
+                    "user": {
+                        "id": "'$USER_ID'",
+                        "passcode": "'$PASSCODE'"
+                    }
+                }
+            }
+        }
+    }' \
+      http://localhost:5000/v3/auth/tokens ; echo

doc/source/index.rst

@@ -55,6 +55,7 @@ Getting Started
     mapping_combinations
     mapping_schema
     configure_tokenless_x509
+    auth-totp
     configuringservices
     extensions
     key_terms

keystone/auth/plugins/core.py

@@ -99,18 +99,17 @@ def convert_integer_to_method_list(method_int):
 
 
 @dependency.requires('identity_api', 'resource_api')
-class UserAuthInfo(object):
+class BaseUserInfo(object):
 
-    @staticmethod
-    def create(auth_payload, method_name):
-        user_auth_info = UserAuthInfo()
+    @classmethod
+    def create(cls, auth_payload, method_name):
+        user_auth_info = cls()
         user_auth_info._validate_and_normalize_auth_data(auth_payload)
         user_auth_info.METHOD_NAME = method_name
         return user_auth_info
 
     def __init__(self):
         self.user_id = None
-        self.password = None
         self.user_ref = None
         self.METHOD_NAME = None
 
@@ -164,7 +163,6 @@ class UserAuthInfo(object):
         if not user_id and not user_name:
             raise exception.ValidationError(attribute='id or name',
                                             target='user')
-        self.password = user_info.get('password')
         try:
             if user_name:
                 if 'domain' not in user_info:
@@ -185,3 +183,29 @@ class UserAuthInfo(object):
         self.user_ref = user_ref
         self.user_id = user_ref['id']
         self.domain_id = domain_ref['id']
+
+
+class UserAuthInfo(BaseUserInfo):
+
+    def __init__(self):
+        super(UserAuthInfo, self).__init__()
+        self.password = None
+
+    def _validate_and_normalize_auth_data(self, auth_payload):
+        super(UserAuthInfo, self)._validate_and_normalize_auth_data(
+            auth_payload)
+        user_info = auth_payload['user']
+        self.password = user_info.get('password')
+
+
+class TOTPUserInfo(BaseUserInfo):
+
+    def __init__(self):
+        super(TOTPUserInfo, self).__init__()
+        self.passcode = None
+
+    def _validate_and_normalize_auth_data(self, auth_payload):
+        super(TOTPUserInfo, self)._validate_and_normalize_auth_data(
+            auth_payload)
+        user_info = auth_payload['user']
+        self.passcode = user_info.get('passcode')

keystone/auth/plugins/totp.py

@@ -0,0 +1,98 @@
+# 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.
+
+"""Time-based One-time Password Algorithm (TOTP) auth plugin
+
+TOTP is an algorithm that computes a one-time password from a shared secret
+key and the current time.
+
+TOTP is an implementation of a hash-based message authentication code (HMAC).
+It combines a secret key with the current timestamp using a cryptographic hash
+function to generate a one-time password. The timestamp typically increases in
+30-second intervals, so passwords generated close together in time from the
+same secret key will be equal.
+"""
+
+import base64
+import time
+
+from cryptography.hazmat.backends import default_backend
+from cryptography.hazmat.primitives import hashes
+from cryptography.hazmat.primitives.twofactor import totp as crypto_totp
+from oslo_log import log
+import six
+
+from keystone import auth
+from keystone.auth import plugins
+from keystone.common import dependency
+from keystone import exception
+from keystone.i18n import _
+
+
+METHOD_NAME = 'totp'
+
+LOG = log.getLogger(__name__)
+
+
+def _get_totp_token(secret):
+    """Generate TOTP code.
+
+    :param bytes secret: A base32 encoded secret for the TOTP authentication
+    :returns: totp passcode as bytes
+    """
+    if isinstance(secret, six.text_type):
+        # NOTE(dstanek): since this may be coming from the JSON stored in the
+        # database it may be UTF-8 encoded
+        secret = secret.encode('utf-8')
+
+    # NOTE(nonameentername): cryptography takes a non base32 encoded value for
+    # TOTP. Add the correct padding to be able to base32 decode
+    while len(secret) % 8 != 0:
+        secret = secret + b'='
+
+    decoded = base64.b32decode(secret)
+    totp = crypto_totp.TOTP(
+        decoded, 6, hashes.SHA1(), 30, backend=default_backend())
+    return totp.generate(time.time())
+
+
+@dependency.requires('credential_api')
+class TOTP(auth.AuthMethodHandler):
+
+    def authenticate(self, context, auth_payload, auth_context):
+        """Try to authenticate using TOTP"""
+        user_info = plugins.TOTPUserInfo.create(auth_payload, METHOD_NAME)
+        auth_passcode = auth_payload.get('user').get('passcode')
+
+        credentials = self.credential_api.list_credentials_for_user(
+            user_info.user_id, type='totp')
+
+        valid_passcode = False
+        for credential in credentials:
+            try:
+                generated_passcode = _get_totp_token(credential['blob'])
+                if auth_passcode == generated_passcode:
+                    valid_passcode = True
+                    break
+            except (ValueError, KeyError):
+                LOG.debug('No TOTP match; credential id: %s, user_id: %s',
+                          credential['id'], user_info.user_id)
+            except (TypeError):
+                LOG.debug('Base32 decode failed for TOTP credential %s',
+                          credential['id'])
+
+        if not valid_passcode:
+            # authentication failed because of invalid username or passcode
+            msg = _('Invalid username or TOTP passcode')
+            raise exception.Unauthorized(msg)
+
+        auth_context['user_id'] = user_info.user_id

keystone/tests/common/auth.py

@@ -45,22 +45,34 @@ class AuthTestMixin(object):
             scope_data['OS-TRUST:trust']['id'] = trust_id
         return scope_data
 
-    def _build_password_auth(self, user_id=None, username=None,
-                             user_domain_id=None, user_domain_name=None,
-                             password=None):
-        password_data = {'user': {}}
+    def _build_auth(self, user_id=None, username=None, user_domain_id=None,
+                    user_domain_name=None, **kwargs):
+
+        # NOTE(dstanek): just to ensure sanity in the tests
+        self.assertEqual(1, len(kwargs),
+                         message='_build_auth requires 1 (and only 1) '
+                                 'secret type and value')
+
+        secret_type, secret_value = kwargs.items()[0]
+
+        # NOTE(dstanek): just to ensure sanity in the tests
+        self.assertIn(secret_type, ('passcode', 'password'),
+                      message="_build_auth only supports 'passcode' "
+                              "and 'password' secret types")
+
+        data = {'user': {}}
         if user_id:
-            password_data['user']['id'] = user_id
+            data['user']['id'] = user_id
         else:
-            password_data['user']['name'] = username
+            data['user']['name'] = username
             if user_domain_id or user_domain_name:
-                password_data['user']['domain'] = {}
+                data['user']['domain'] = {}
                 if user_domain_id:
-                    password_data['user']['domain']['id'] = user_domain_id
+                    data['user']['domain']['id'] = user_domain_id
                 else:
-                    password_data['user']['domain']['name'] = user_domain_name
-        password_data['user']['password'] = password
-        return password_data
+                    data['user']['domain']['name'] = user_domain_name
+        data['user'][secret_type] = secret_value
+        return data
 
     def _build_token_auth(self, token):
         return {'id': token}
@@ -68,7 +80,7 @@ class AuthTestMixin(object):
     def build_authentication_request(self, token=None, user_id=None,
                                      username=None, user_domain_id=None,
                                      user_domain_name=None, password=None,
-                                     kerberos=False, **kwargs):
+                                     kerberos=False, passcode=None, **kwargs):
         """Build auth dictionary.
 
         It will create an auth dictionary based on all the arguments
@@ -82,10 +94,16 @@ class AuthTestMixin(object):
         if token:
             auth_data['identity']['methods'].append('token')
             auth_data['identity']['token'] = self._build_token_auth(token)
-        if user_id or username:
+        if password and (user_id or username):
             auth_data['identity']['methods'].append('password')
-            auth_data['identity']['password'] = self._build_password_auth(
-                user_id, username, user_domain_id, user_domain_name, password)
+            auth_data['identity']['password'] = self._build_auth(
+                user_id, username, user_domain_id, user_domain_name,
+                password=password)
+        if passcode and (user_id or username):
+            auth_data['identity']['methods'].append('totp')
+            auth_data['identity']['totp'] = self._build_auth(
+                user_id, username, user_domain_id, user_domain_name,
+                passcode=passcode)
         if kwargs:
             auth_data['scope'] = self._build_auth_scope(**kwargs)
         return {'auth': auth_data}

keystone/tests/unit/core.py

@@ -14,6 +14,7 @@
 
 from __future__ import absolute_import
 import atexit
+import base64
 import datetime
 import functools
 import hashlib
@@ -398,6 +399,16 @@ def new_ec2_credential(user_id, project_id=None, blob=None, **kwargs):
     return blob, credential
 
 
+def new_totp_credential(user_id, project_id=None, blob=None):
+    if not blob:
+        blob = base64.b32encode(uuid.uuid4().hex).rstrip('=')
+    credential = new_credential_ref(user_id=user_id,
+                                    project_id=project_id,
+                                    blob=blob,
+                                    type='totp')
+    return credential
+
+
 def new_role_ref(**kwargs):
     ref = {
         'id': uuid.uuid4().hex,

keystone/tests/unit/test_v3_auth.py

@@ -14,6 +14,7 @@
 
 import copy
 import datetime
+import itertools
 import json
 import operator
 import uuid
@@ -29,6 +30,7 @@ from testtools import matchers
 from testtools import testcase
 
 from keystone import auth
+from keystone.auth.plugins import totp
 from keystone.common import utils
 from keystone.contrib.revoke import routers
 from keystone import exception
@@ -4658,3 +4660,149 @@ class TestAuthFernetTokenProvider(TestAuth):
         # Bind not current supported by Fernet, see bug 1433311.
         self.v3_create_token(auth_data,
                              expected_status=http_client.NOT_IMPLEMENTED)
+
+
+class TestAuthTOTP(test_v3.RestfulTestCase):
+
+    def setUp(self):
+        super(TestAuthTOTP, self).setUp()
+
+        ref = unit.new_totp_credential(
+            user_id=self.default_domain_user['id'],
+            project_id=self.default_domain_project['id'])
+
+        self.secret = ref['blob']
+
+        r = self.post('/credentials', body={'credential': ref})
+        self.assertValidCredentialResponse(r, ref)
+
+        self.addCleanup(self.cleanup)
+
+    def auth_plugin_config_override(self):
+        methods = ['totp', 'token', 'password']
+        super(TestAuthTOTP, self).auth_plugin_config_override(methods)
+
+    def _make_credentials(self, cred_type, count=1, user_id=None,
+                          project_id=None, blob=None):
+        user_id = user_id or self.default_domain_user['id']
+        project_id = project_id or self.default_domain_project['id']
+
+        creds = []
+        for __ in range(count):
+            if cred_type == 'totp':
+                ref = unit.new_totp_credential(
+                    user_id=user_id, project_id=project_id, blob=blob)
+            else:
+                ref = unit.new_credential_ref(
+                    user_id=user_id, project_id=project_id)
+            resp = self.post('/credentials', body={'credential': ref})
+            creds.append(resp.json['credential'])
+        return creds
+
+    def _make_auth_data_by_id(self, passcode, user_id=None):
+        return self.build_authentication_request(
+            user_id=user_id or self.default_domain_user['id'],
+            passcode=passcode,
+            project_id=self.project['id'])
+
+    def _make_auth_data_by_name(self, passcode, username, user_domain_id):
+        return self.build_authentication_request(
+            username=username,
+            user_domain_id=user_domain_id,
+            passcode=passcode,
+            project_id=self.project['id'])
+
+    def cleanup(self):
+        totp_creds = self.credential_api.list_credentials_for_user(
+            self.default_domain_user['id'], type='totp')
+
+        other_creds = self.credential_api.list_credentials_for_user(
+            self.default_domain_user['id'], type='other')
+
+        for cred in itertools.chain(other_creds, totp_creds):
+            self.delete('/credentials/%s' % cred['id'],
+                        expected_status=http_client.NO_CONTENT)
+
+    def test_with_a_valid_passcode(self):
+        creds = self._make_credentials('totp')
+        secret = creds[-1]['blob']
+        auth_data = self._make_auth_data_by_id(totp._get_totp_token(secret))
+
+        self.v3_create_token(auth_data, expected_status=http_client.CREATED)
+
+    def test_with_an_invalid_passcode_and_user_credentials(self):
+        self._make_credentials('totp')
+        auth_data = self._make_auth_data_by_id('000000')
+        self.v3_create_token(auth_data,
+                             expected_status=http_client.UNAUTHORIZED)
+
+    def test_with_an_invalid_passcode_with_no_user_credentials(self):
+        auth_data = self._make_auth_data_by_id('000000')
+        self.v3_create_token(auth_data,
+                             expected_status=http_client.UNAUTHORIZED)
+
+    def test_with_a_corrupt_totp_credential(self):
+        self._make_credentials('totp', count=1, blob='0')
+        auth_data = self._make_auth_data_by_id('000000')
+        self.v3_create_token(auth_data,
+                             expected_status=http_client.UNAUTHORIZED)
+
+    def test_with_multiple_credentials(self):
+        self._make_credentials('other', 3)
+        creds = self._make_credentials('totp', count=3)
+        secret = creds[-1]['blob']
+
+        auth_data = self._make_auth_data_by_id(totp._get_totp_token(secret))
+        self.v3_create_token(auth_data, expected_status=http_client.CREATED)
+
+    def test_with_multiple_users(self):
+        # make some credentials for the existing user
+        self._make_credentials('totp', count=3)
+
+        # create a new user and their credentials
+        user = unit.create_user(self.identity_api, domain_id=self.domain_id)
+        self.assignment_api.create_grant(self.role['id'],
+                                         user_id=user['id'],
+                                         project_id=self.project['id'])
+        creds = self._make_credentials('totp', count=1, user_id=user['id'])
+        secret = creds[-1]['blob']
+
+        auth_data = self._make_auth_data_by_id(
+            totp._get_totp_token(secret), user_id=user['id'])
+        self.v3_create_token(auth_data, expected_status=http_client.CREATED)
+
+    def test_with_multiple_users_and_invalid_credentials(self):
+        """Prevent logging in with someone else's credentials.
+
+        It's very easy to forget to limit the credentials query by user.
+        Let's just test it for a sanity check.
+        """
+        # make some credentials for the existing user
+        self._make_credentials('totp', count=3)
+
+        # create a new user and their credentials
+        new_user = unit.create_user(self.identity_api,
+                                    domain_id=self.domain_id)
+        self.assignment_api.create_grant(self.role['id'],
+                                         user_id=new_user['id'],
+                                         project_id=self.project['id'])
+        user2_creds = self._make_credentials(
+            'totp', count=1, user_id=new_user['id'])
+
+        user_id = self.default_domain_user['id']  # user1
+        secret = user2_creds[-1]['blob']
+
+        auth_data = self._make_auth_data_by_id(
+            totp._get_totp_token(secret), user_id=user_id)
+        self.v3_create_token(auth_data,
+                             expected_status=http_client.UNAUTHORIZED)
+
+    def test_with_username_and_domain_id(self):
+        creds = self._make_credentials('totp')
+        secret = creds[-1]['blob']
+        auth_data = self._make_auth_data_by_name(
+            totp._get_totp_token(secret),
+            username=self.default_domain_user['name'],
+            user_domain_id=self.default_domain_user['domain_id'])
+
+        self.v3_create_token(auth_data, expected_status=http_client.CREATED)

releasenotes/notes/totp-40d93231714c6a20.yaml

@@ -0,0 +1,8 @@
+---
+features:
+  - >
+   [`blueprint totp-auth <https://blueprints.launchpad.net/keystone/+spec/totp-auth>`_]
+   Keystone now supports authenticating via Time-based One-time Password (TOTP).
+   To enable this feature, add the ``totp`` auth plugin to the ``methods``
+   option in the ``[auth]`` section of ``keystone.conf``. More information
+   about using TOTP can be found in keystone's documentation.

setup.cfg

@@ -101,6 +101,9 @@ keystone.auth.saml2 =
 keystone.auth.token =
     default = keystone.auth.plugins.token:Token
 
+keystone.auth.totp =
+    default = keystone.auth.plugins.totp:TOTP
+
 keystone.auth.x509 =
     default = keystone.auth.plugins.mapped:Mapped