Merge remote-tracking branch 'fspoettel/admin-panel-2fa' into totp

# Conflicts: # management/auth.py # management/daemon.py # setup/mail-users.sh # setup/management.sh # setup/migrate.py
2026-03-05 15:57:23 +01:00 · 2020-09-10 15:23:27 -04:00
parent b10f82152a 2ea97f0643
commit 24ae913d68
13 changed files with 718 additions and 39 deletions
--- a/management/auth.py
+++ b/management/auth.py
@@ -2,7 +2,7 @@ import base64, os, os.path, hmac

 from flask import make_response

-import utils
+import utils, totp
 from mailconfig import validate_login, get_mail_password, get_mail_user_privileges

 DEFAULT_KEY_PATH   = '/var/lib/mailinabox/api.key'
@@ -76,23 +76,36 @@ class KeyAuthService:
 			return (None, ["admin"])
 		else:
 			# The user is trying to log in with a username and user-specific
-			# API key or password. Raises or returns privs.
-			return (username, self.get_user_credentials(username, password, env))
+			# API key or password. Raises or returns privs and an indicator
+			# whether the user is using their password or a user-specific API-key.
+			privs, is_user_key = self.get_user_credentials(username, password, env)
+
+			# If the user is using their API key to login, 2FA has been passed before
+			if is_user_key:
+				return (username, privs)
+
+			totp_strategy = totp.TOTPStrategy(email=username)
+			# this will raise `totp.MissingTokenError` or `totp.BadTokenError` for bad requests
+			totp_strategy.validate_request(request, env)
+
+			return (username, privs)

 	def get_user_credentials(self, email, pw, env):
 		# Validate a user's credentials. On success returns a list of
 		# privileges (e.g. [] or ['admin']). On failure raises a ValueError
-		# with a login error message. 
+		# with a login error message.

 		# Sanity check.
 		if email == "" or pw == "":
 			raise ValueError("Enter an email address and password.")

+		is_user_key = False
+
 		# The password might be a user-specific API key. create_user_key raises
 		# a ValueError if the user does not exist.
 		if hmac.compare_digest(self.create_user_key(email, env), pw):
 			# OK.
-			pass
+			is_user_key = True
 		else:
 			# Get the hashed password of the user. Raise a ValueError if the
 			# email address does not correspond to a user.
@@ -107,7 +120,7 @@ class KeyAuthService:
 		if isinstance(privs, tuple): raise ValueError(privs[0])

 		# Return a list of privileges.
-		return privs
+		return (privs, is_user_key)

 	def create_user_key(self, email, env):
 		# Store an HMAC with the client. The hashed message of the HMAC will be the user's
--- a/management/backend.py
+++ b/management/backend.py
@@ -192,7 +192,7 @@ class LdapConnection(ldap3.Connection):
 		#         have values for each attribute in `attrs_to_update`
 		#   attrs_to_update: an array of attribute names to update
 		#   objectClasses: a list of object classes for a new entry
-		#   values: a dict of attributes and values for a new entry
+		#   values: a dict of attributes and values for a new or modified entry
 		if existing_record:
 			# modify existing
 			changes = {}
--- a/management/daemon.py
+++ b/management/daemon.py
@@ -1,14 +1,15 @@
 import os, os.path, re, json, time
-import subprocess
+import multiprocessing.pool, subprocess

 from functools import wraps

 from flask import Flask, request, render_template, abort, Response, send_from_directory, make_response

-import auth, utils, multiprocessing.pool
+import auth, utils, totp
 from mailconfig import get_mail_users, get_mail_users_ex, get_admins, add_mail_user, set_mail_password, set_mail_display_name, remove_mail_user
 from mailconfig import get_mail_user_privileges, add_remove_mail_user_privilege
 from mailconfig import get_mail_aliases, get_mail_aliases_ex, get_mail_domains, add_mail_alias, remove_mail_alias
+from mailconfig import get_mfa_state, create_totp_credential, delete_totp_credential

 env = utils.load_environment()

@@ -37,15 +38,22 @@ def authorized_personnel_only(viewfunc):
 	def newview(*args, **kwargs):
 		# Authenticate the passed credentials, which is either the API key or a username:password pair.
 		error = None
+		privs = []
+
 		try:
 			email, privs = auth_service.authenticate(request, env)
-		except ValueError as e:
-			# Authentication failed.
-			privs = []
-			error = "Incorrect username or password"

+		except totp.MissingTokenError as e:
+			error = str(e)
+		except totp.BadTokenError as e:
 			# Write a line in the log recording the failed login
 			log_failed_login(request)
+			error = str(e)
+		except ValueError as e:
+			# Write a line in the log recording the failed login
+			log_failed_login(request)
+			# Authentication failed.
+			error = "Incorrect username or password"

 		# Authorized to access an API view?
 		if "admin" in privs:
@@ -83,8 +91,8 @@ def authorized_personnel_only(viewfunc):
 def unauthorized(error):
 	return auth_service.make_unauthorized_response()

-def json_response(data):
-	return Response(json.dumps(data, indent=2, sort_keys=True)+'\n', status=200, mimetype='application/json')
+def json_response(data, status=200):
+	return Response(json.dumps(data, indent=2, sort_keys=True)+'\n', status=status, mimetype='application/json')

 ###################################

@@ -118,6 +126,20 @@ def me():
 	# Is the caller authorized?
 	try:
 		email, privs = auth_service.authenticate(request, env)
+	except totp.MissingTokenError as e:
+		return json_response({
+			"status": "missing_token",
+			"reason": str(e),
+		})
+	except totp.BadTokenError as e:
+		# Log the failed login
+		log_failed_login(request)
+
+		return json_response({
+			"status": "bad_token",
+			"reason": str(e),
+		})
+
 	except ValueError as e:
 		# Log the failed login
 		log_failed_login(request)
@@ -125,7 +147,7 @@ def me():
 		return json_response({
 			"status": "invalid",
 			"reason": "Incorrect username or password",
-			})
+		})

 	resp = {
 		"status": "ok",
@@ -343,7 +365,7 @@ def ssl_get_status():

 	# What domains can we provision certificates for? What unexpected problems do we have?
 	provision, cant_provision = get_certificates_to_provision(env, show_valid_certs=False)
-	
+
 	# What's the current status of TLS certificates on all of the domain?
 	domains_status = get_web_domains_info(env)
 	domains_status = [
@@ -392,6 +414,51 @@ def ssl_provision_certs():
 	requests = provision_certificates(env, limit_domains=None)
 	return json_response({ "requests": requests })

+# multi-factor auth
+
+@app.route('/mfa/status', methods=['GET'])
+@authorized_personnel_only
+def two_factor_auth_get_status():
+	email, _ = auth_service.authenticate(request, env)
+
+	mfa_state = get_mfa_state(email, env)
+
+	if mfa_state['type'] == 'totp':
+		return json_response({ "type": 'totp' })
+
+	secret = totp.get_secret()
+	secret_url = totp.get_otp_uri(secret, email)
+	secret_qr = totp.get_qr_code(secret_url)
+
+	return json_response({
+		"type": None,
+		"totp_secret": secret,
+		"totp_qr": secret_qr
+	})
+
+@app.route('/mfa/totp/enable', methods=['POST'])
+@authorized_personnel_only
+def totp_post_enable():
+	email, _ = auth_service.authenticate(request, env)
+
+	secret = request.form.get('secret')
+	token = request.form.get('token')
+
+	if type(secret) != str or type(token) != str or len(token) != 6 or len(secret) != 32:
+		return json_response({ "error": 'bad_input' }, 400)
+
+	if totp.validate(secret, token):
+		create_totp_credential(email, secret, token, env)
+		return json_response({})
+
+	return json_response({ "error": 'token_mismatch' }, 400)
+
+@app.route('/mfa/totp/disable', methods=['POST'])
+@authorized_personnel_only
+def totp_post_disable():
+	email, _ = auth_service.authenticate(request, env)
+	delete_totp_credential(email, env)
+	return json_response({})

 # WEB

--- a/management/mailconfig.py
+++ b/management/mailconfig.py
@@ -1129,6 +1129,76 @@ def get_required_aliases(env):

 	return aliases

+# multi-factor auth
+
+def get_mfa_state(email, env):
+	# find the user				  
+	conn = open_database(env)
+	user = find_mail_user(env, email, ['objectClass','totpSecret','totpMruToken'], conn)
+	if user is None or 'totpUser' not in user['objectClass']:
+		return { 'type': None }
+
+	secret = user['totpSecret'][0]
+	mru_token = None
+	if len(user['totpMruToken'])>0:
+		mru_token = user['totpMruToken'][0]
+
+	return {
+		'type': 'totp',
+		'secret': secret,
+		'mru_token': '' if mru_token is None else mru_token
+	}
+
+def create_totp_credential(email, secret, token, env):
+	validate_totp_secret(secret)
+	conn = open_database(env)
+	user = find_mail_user(env, email, ['objectClass','totpSecret','totpMruToken'], conn)
+	if user is None:
+		return ("That's not a user (%s)." % email, 400)
+
+	attrs = {
+		"totpSecret": secret,
+		"totpMruToken": token
+	}
+	if 'totpUser' not in user['objectClass']:
+		attrs['objectClass'] = user['objectClass'].copy()
+		attrs['objectClass'].append('totpUser')		
+	conn.add_or_modify(user['dn'], user, attrs.keys(), None, attrs)
+	return "OK"
+
+def set_mru_totp_code(email, token, env):
+	conn = open_database(env)
+	user = find_mail_user(env, email, ['objectClass','totpMruToken'], conn)
+	if user is None:
+		return ("That's not a user (%s)." % email, 400)
+	
+	if 'totpUser' not in user['objectClass']:
+		return ("User (%s) not configured for TOTP" % email, 400)
+
+	attrs = {
+		"totpMruToken": token
+	}			
+	conn.add_or_modify(user['dn'], user, attrs.keys(), None, attrs)
+	return "OK"
+
+def delete_totp_credential(email, env):
+	conn = open_database(env)
+	user = find_mail_user(env, email, ['objectClass','totpSecret','totpMruToken'], conn)
+	if user is None:
+		return ("That's not a user (%s)." % email, 400)
+	
+	if 'totpUser' not in user['objectClass']:
+		return "OK"
+
+	attrs = {
+		"totpMruToken": None,
+		"totpSecret": None,
+		"objectClass": user["objectClass"].copy()
+	}
+	attrs["objectClass"].remove("totpUser")	
+	conn.add_or_modify(user['dn'], user, attrs.keys(), None, attrs)
+	return "OK"
+
 def kick(env, mail_result=None):
 	results = []

@@ -1190,6 +1260,11 @@ def validate_password(pw):
 	if len(pw) < 8:
 		raise ValueError("Passwords must be at least eight characters.")

+def validate_totp_secret(secret):
+	if type(secret) != str or secret.strip() == "":
+		raise ValueError("No secret provided.")
+	if len(secret) != 32:
+		raise ValueError("Secret should be a 32 characters base32 string")

 if __name__ == "__main__":
 	import sys
--- a/management/templates/index.html
+++ b/management/templates/index.html
@@ -93,6 +93,7 @@
                <li class="dropdown-header">Advanced Pages</li>
                <li><a href="#custom_dns" onclick="return show_panel(this);">Custom DNS</a></li>
                <li><a href="#external_dns" onclick="return show_panel(this);">External DNS</a></li>
+                <li><a href="#two_factor_auth" onclick="return show_panel(this);">Two-Factor Authentication</a></li>
                <li><a href="/admin/munin" target="_blank">Munin Monitoring</a></li>
              </ul>
            </li>
@@ -131,6 +132,10 @@
      {% include "custom-dns.html" %}
      </div>

+      <div id="panel_two_factor_auth" class="admin_panel">
+      {% include "two-factor-auth.html" %}
+      </div>
+
      <div id="panel_login" class="admin_panel">
      {% include "login.html" %}
      </div>
@@ -292,7 +297,7 @@ function ajax_with_indicator(options) {
 }

 var api_credentials = ["", ""];
-function api(url, method, data, callback, callback_error) {
+function api(url, method, data, callback, callback_error, headers) {
  // from http://www.webtoolkit.info/javascript-base64.html
  function base64encode(input) {
    _keyStr = "ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz0123456789+/=";
@@ -330,7 +335,7 @@ function api(url, method, data, callback, callback_error) {
    method: method,
    cache: false,
    data: data,
-
+    headers: headers,
    // the custom DNS api sends raw POST/PUT bodies --- prevent URL-encoding
    processData: typeof data != "string",
    mimeType: typeof data == "string" ? "text/plain; charset=ascii" : null,
--- a/management/templates/login.html
+++ b/management/templates/login.html
@@ -1,4 +1,29 @@
-<h1 style="margin: 1em; text-align: center">{{hostname}}</h1>
+<style>
+  .title {
+    margin: 1em;
+    text-align: center;
+  }
+
+  .subtitle {
+    margin: 2em;
+    text-align: center;
+  }
+
+  .login {
+    margin: 0 auto;
+    max-width: 32em;
+  }
+
+  .login #loginOtp {
+    display: none;
+  }
+
+  #loginForm.is-twofactor #loginOtp {
+    display: block
+  }
+</style>
+
+<h1 class="title">{{hostname}}</h1>

 {% if no_users_exist or no_admins_exist %}
 <div class="row">
@@ -20,10 +45,10 @@ sudo tools/mail.py user make-admin me@{{hostname}}</pre>
 </div>
 {% endif %}

-<p style="margin: 2em; text-align: center;">Log in here for your Mail-in-a-Box control panel.</p>
+<p class="subtitle">Log in here for your Mail-in-a-Box control panel.</p>

-<div style="margin: 0 auto; max-width: 32em;">
-  <form class="form-horizontal" role="form" onsubmit="do_login(); return false;" method="get">
+<div class="login">
+  <form id="loginForm" class="form-horizontal" role="form" onsubmit="do_login(); return false;" method="get">
    <div class="form-group">
      <label for="inputEmail3" class="col-sm-3 control-label">Email</label>
      <div class="col-sm-9">
@@ -45,6 +70,12 @@ sudo tools/mail.py user make-admin me@{{hostname}}</pre>
        </div>
      </div>
    </div>
+    <div class="form-group" id="loginOtp">
+        <div class="col-sm-offset-3 col-sm-9">
+          <label for="loginOtpInput" class="control-label">Two-Factor Code</label>
+          <input type="text" class="form-control" id="loginOtpInput" placeholder="6-digit code">
+        </div>
+    </div>
    <div class="form-group">
      <div class="col-sm-offset-3 col-sm-9">
        <button type="submit" class="btn btn-default">Sign in</button>
@@ -53,15 +84,15 @@ sudo tools/mail.py user make-admin me@{{hostname}}</pre>
  </form>
 </div>

-
 <script>
 function do_login() {
  if ($('#loginEmail').val() == "") {
    show_modal_error("Login Failed", "Enter your email address.", function() {
-	$('#loginEmail').focus();
+      $('#loginEmail').focus();
    });
    return false;
  }
+
  if ($('#loginPassword').val() == "") {
    show_modal_error("Login Failed", "Enter your email password.", function() {
        $('#loginPassword').focus();
@@ -75,17 +106,24 @@ function do_login() {
  api(
  "/me",
  "GET",
-  { },
-  function(response){
+  {},
+  function(response) {
    // This API call always succeeds. It returns a JSON object indicating
    // whether the request was authenticated or not.
-    if (response.status != "ok") {
-      // Show why the login failed.
-      show_modal_error("Login Failed", response.reason)
-
-      // Reset any saved credentials.
-      do_logout();
+    if (response.status != 'ok') {
+      if (response.status === 'missing_token' && !$('#loginForm').hasClass('is-twofactor')) {
+        $('#loginForm').addClass('is-twofactor');
+        setTimeout(() => {
+            $('#loginOtpInput').focus();
+        });
+      } else {
+        $('#loginForm').removeClass('is-twofactor');
+        // Show why the login failed.
+        show_modal_error("Login Failed", response.reason)

+        // Reset any saved credentials.
+        do_logout();
+      }
    } else if (!("api_key" in response)) {
      // Login succeeded but user might not be authorized!
      show_modal_error("Login Failed", "You are not an administrator on this system.")
@@ -102,6 +140,8 @@ function do_login() {
      // Try to wipe the username/password information.
      $('#loginEmail').val('');
      $('#loginPassword').val('');
+      $('#loginOtpInput').val('');
+      $('#loginForm').removeClass('is-twofactor');

      // Remember the credentials.
      if (typeof localStorage != 'undefined' && typeof sessionStorage != 'undefined') {
@@ -119,7 +159,11 @@ function do_login() {
      // which confuses the loading indicator.
      setTimeout(function() { show_panel(!switch_back_to_panel || switch_back_to_panel == "login" ? 'system_status' : switch_back_to_panel) }, 300);
    }
-  })
+  },
+  undefined,
+  {
+    'x-auth-token': $('#loginOtpInput').val()
+  });
 }

 function do_logout() {
@@ -132,6 +176,8 @@ function do_logout() {
 }

 function show_login() {
+  $('#loginForm').removeClass('is-twofactor');
+  $('#loginOtpInput').val('');
  $('#loginEmail,#loginPassword').each(function() {
    var input = $(this);
    if (!$.trim(input.val())) {
--- a/management/templates/two-factor-auth.html
+++ b/management/templates/two-factor-auth.html
@@ -0,0 +1,220 @@
+<style>
+    .twofactor #totp-setup,
+    .twofactor #disable-2fa,
+    .twofactor #output-2fa {
+        display: none;
+    }
+
+    .twofactor.loaded .loading-indicator {
+        display: none;
+    }
+
+    .twofactor.disabled #disable-2fa,
+    .twofactor.enabled #totp-setup {
+        display: none;
+    }
+
+    .twofactor.disabled #totp-setup,
+    .twofactor.enabled #disable-2fa {
+        display: block;
+    }
+
+    .twofactor #totp-setup-qr img {
+        display: block;
+        width: 256px;
+        max-width: 100%;
+        height: auto;
+    }
+
+    .twofactor #output-2fa.visible {
+        display: block;
+    }
+</style>
+
+<h2>Two-Factor Authentication</h2>
+
+<div class="twofactor">
+    <div class="loading-indicator">Loading...</div>
+
+    <form id="totp-setup">
+        <div class="form-group">
+            <h3>Setup</h3>
+            <p>After enabling two factor authentication, any login to the admin panel will require you to enter a time-limited 6-digit number from an authenticator app after entering your normal credentials.</p>
+            <p>1. Scan the QR code or enter the secret into an authenticator app (e.g. Google Authenticator)</p>
+            <div id="totp-setup-qr"></div>
+        </div>
+
+        <div class="form-group">
+            <label for="otp">2. Enter the code displayed in the Authenticator app</label>
+            <input type="text" id="totp-setup-token" class="form-control" placeholder="6-digit code" />
+        </div>
+
+        <input type="hidden" id="totp-setup-secret" />
+
+        <div class="form-group">
+            <button id="totp-setup-submit" disabled type="submit" class="btn">Enable two factor authentication</button>
+        </div>
+    </form>
+
+    <form id="disable-2fa">
+        <div class="form-group">
+            <p>Two factor authentication is active.</p>
+        </div>
+
+        <button type="submit" class="btn btn-danger">Disable two factor authentication</button>
+    </form>
+
+    <div id="output-2fa" class="panel panel-danger">
+        <div class="panel-body"></div>
+    </div>
+</div>
+
+<script>
+    var el = {
+        disableForm: document.getElementById('disable-2fa'),
+        output: document.getElementById('output-2fa'),
+        totpSetupForm: document.getElementById('totp-setup'),
+        totpSetupToken: document.getElementById('totp-setup-token'),
+        totpSetupSecret: document.getElementById('totp-setup-secret'),
+        totpQr: document.getElementById('totp-setup-qr'),
+        totpSetupSubmit: document.querySelector('#totp-setup-submit'),
+        wrapper: document.querySelector('.twofactor')
+    }
+
+    function update_setup_disabled(evt) {
+        var val = evt.target.value.trim();
+
+        if (
+            typeof val !== 'string' ||
+            typeof el.totpSetupSecret.value !== 'string' ||
+            val.length !== 6 ||
+            el.totpSetupSecret.value.length !== 32 ||
+            !(/^\+?\d+$/.test(val))
+        ) {
+            el.totpSetupSubmit.setAttribute('disabled', '');
+        } else {
+            el.totpSetupSubmit.removeAttribute('disabled');
+        }
+    }
+
+    function render_totp_setup(res) {
+        function render_qr_code(encoded) {
+            var img = document.createElement('img');
+            img.src = encoded;
+
+            var code = document.createElement('div');
+            code.innerHTML = `Secret: ${res.totp_secret}`;
+
+            el.totpQr.appendChild(img);
+            el.totpQr.appendChild(code);
+        }
+
+        el.totpSetupToken.addEventListener('input', update_setup_disabled);
+        el.totpSetupForm.addEventListener('submit', do_enable_totp);
+
+        el.totpSetupSecret.setAttribute('value', res.totp_secret);
+        render_qr_code(res.totp_qr);
+
+        el.wrapper.classList.add('disabled');
+    }
+
+    function render_disable() {
+        el.disableForm.addEventListener('submit', do_disable);
+        el.wrapper.classList.add('enabled');
+    }
+
+    function hide_error() {
+        el.output.querySelector('.panel-body').innerHTML = '';
+        el.output.classList.remove('visible');
+    }
+
+    function render_error(msg) {
+        el.output.querySelector('.panel-body').innerHTML = msg;
+        el.output.classList.add('visible');
+    }
+
+    function reset_view() {
+        el.wrapper.classList.remove('loaded', 'disabled', 'enabled');
+
+        el.disableForm.removeEventListener('submit', do_disable);
+
+        hide_error();
+
+        el.totpSetupForm.reset();
+        el.totpSetupForm.removeEventListener('submit', do_enable_totp);
+
+        el.totpSetupSecret.setAttribute('value', '');
+        el.totpSetupToken.removeEventListener('input', update_setup_disabled);
+
+        el.totpSetupSubmit.setAttribute('disabled', '');
+        el.totpQr.innerHTML = '';
+    }
+
+    function show_two_factor_auth() {
+        reset_view();
+
+        api(
+            '/mfa/status',
+            'GET',
+            {},
+            function(res) {
+                el.wrapper.classList.add('loaded');
+                var isTotpEnabled = res.type === 'totp'
+                return isTotpEnabled ? render_disable(res) : render_totp_setup(res);
+            }
+        );
+    }
+
+    function do_disable(evt) {
+        evt.preventDefault();
+        hide_error();
+
+        api(
+            '/mfa/totp/disable',
+            'POST',
+            {},
+            function() { show_two_factor_auth(); }
+        );
+
+        return false;
+    }
+
+    function do_enable_totp(evt) {
+        evt.preventDefault();
+        hide_error();
+
+        api(
+            '/mfa/totp/enable',
+            'POST',
+            {
+                token: $(el.totpSetupToken).val(),
+                secret: $(el.totpSetupSecret).val()
+            },
+            function(res) { show_two_factor_auth(); },
+            function(res) {
+                var errorMessage = 'Something went wrong.';
+                var parsed;
+
+                try {
+                    parsed = JSON.parse(res);
+                } catch (err) {
+                    return render_error(errorMessage);
+                }
+
+                var error = parsed && parsed.error
+                    ? parsed.error
+                    : null;
+
+                if (error === 'token_mismatch') {
+                    errorMessage = 'Code does not match.';
+                } else if (error === 'bad_input') {
+                    errorMessage = 'Received request with malformed data.';
+                }
+
+                render_error(errorMessage);
+            }
+        );
+
+        return false;
+    }
+</script>
--- a/management/totp.py
+++ b/management/totp.py
@@ -0,0 +1,72 @@
+import base64
+import hmac
+import io
+import os
+import struct
+import time
+import pyotp
+import qrcode
+from mailconfig import get_mfa_state, set_mru_totp_code
+
+def get_secret():
+	return base64.b32encode(os.urandom(20)).decode('utf-8')
+
+def get_otp_uri(secret, email):
+	return pyotp.TOTP(secret).provisioning_uri(
+		name=email,
+		issuer_name='mailinabox'
+	)
+
+def get_qr_code(data):
+	qr = qrcode.make(data)
+	byte_arr = io.BytesIO()
+	qr.save(byte_arr, format='PNG')
+
+	encoded = base64.b64encode(byte_arr.getvalue()).decode('utf-8')
+	return 'data:image/png;base64,{}'.format(encoded)
+
+def validate(secret, token):
+	"""
+	@see https://tools.ietf.org/html/rfc6238#section-4
+	@see https://tools.ietf.org/html/rfc4226#section-5.4
+	"""
+	totp = pyotp.TOTP(secret)
+	return totp.verify(token, valid_window=1)
+
+class MissingTokenError(ValueError):
+	pass
+
+class BadTokenError(ValueError):
+	pass
+
+class TOTPStrategy():
+	def __init__(self, email):
+		self.type = 'totp'
+		self.email = email
+
+	def store_successful_login(self, token, env):
+		return set_mru_totp_code(self.email, token, env)
+
+	def validate_request(self, request, env):
+		mfa_state = get_mfa_state(self.email, env)
+
+		# 2FA is not enabled, we can skip further checks
+		if mfa_state['type'] != 'totp':
+			return True
+
+		# If 2FA is enabled, raise if:
+		# 1. no token is provided via `x-auth-token`
+		# 2. a previously supplied token is used (to counter replay attacks)
+		# 3. the token is invalid
+		# in that case, we need to raise and indicate to the client to supply a TOTP
+		token_header = request.headers.get('x-auth-token')
+
+		if not token_header:
+			raise MissingTokenError("Two factor code missing (no x-auth-token supplied)")
+
+		# TODO: Should a token replay be handled as its own error?
+		if hmac.compare_digest(token_header, mfa_state['mru_token']) or validate(mfa_state['secret'], token_header) != True:
+			raise BadTokenError("Two factor code incorrect")
+
+		self.store_successful_login(token_header, env)
+		return True