WebAuthn MFA for the control panel
This commit is contained in:
Joshua Tauberer
parent
30f067bc72
commit
d4428e1c67
|
|
@ -21,7 +21,7 @@ import auth, utils
|
|||
from mailconfig import get_mail_users, get_mail_users_ex, get_admins, add_mail_user, set_mail_password, 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 mfa import get_public_mfa_state, provision_totp, validate_totp_secret, enable_mfa, disable_mfa
|
||||
from mfa import get_public_mfa_state, provision_totp, provision_webauthn, validate_totp_secret, enable_mfa, disable_mfa
|
||||
|
||||
env = utils.load_environment()
|
||||
|
||||
|
|
@ -478,7 +478,8 @@ def mfa_get_status():
|
|||
if email == request.user_email:
|
||||
resp.update({
|
||||
"new_mfa": {
|
||||
"totp": provision_totp(email, env)
|
||||
"totp": provision_totp(email, env),
|
||||
"webauthn": provision_webauthn(email, env)
|
||||
}
|
||||
})
|
||||
except ValueError as e:
|
||||
|
|
@ -495,7 +496,20 @@ def totp_post_enable():
|
|||
return ("Bad Input", 400)
|
||||
try:
|
||||
validate_totp_secret(secret)
|
||||
enable_mfa(request.user_email, "totp", secret, token, label, env)
|
||||
enable_mfa(request.user_email, "totp", env, secret, token, label)
|
||||
except ValueError as e:
|
||||
return (str(e), 400)
|
||||
return "OK"
|
||||
|
||||
@app.route('/mfa/enable/webauthn', methods=['POST'])
|
||||
@authorized_personnel_only
|
||||
def webauthn_post_enable():
|
||||
attestationObject = request.form.get('attestationObject')
|
||||
clientDataJSON = request.form.get('clientDataJSON')
|
||||
if type(attestationObject) != str or type(clientDataJSON) != str:
|
||||
return ("Bad Input", 400)
|
||||
try:
|
||||
enable_mfa(request.user_email, "webauthn", env, attestationObject, clientDataJSON)
|
||||
except ValueError as e:
|
||||
return (str(e), 400)
|
||||
return "OK"
|
||||
|
|
|
|||
|
|
@ -1,9 +1,12 @@
|
|||
import base64
|
||||
import hmac
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
import pyotp
|
||||
import qrcode
|
||||
import pywarp
|
||||
import pywarp.backends
|
||||
|
||||
from mailconfig import open_database
|
||||
|
||||
|
|
@ -29,25 +32,44 @@ def get_public_mfa_state(email, env):
|
|||
]
|
||||
|
||||
def get_hash_mfa_state(email, env):
|
||||
mfa_state = get_mfa_state(email, env)
|
||||
return [
|
||||
{ "id": s["id"], "type": s["type"], "secret": s["secret"] }
|
||||
for s in mfa_state
|
||||
]
|
||||
# Get the current MFA credential secrets from which we form a hash
|
||||
# so that we can reset user logins when any authentication information
|
||||
# changes.
|
||||
mfa_state = []
|
||||
for s in get_mfa_state(email, env):
|
||||
# Add TOTP id and secret to the state.
|
||||
# Skip WebAuthn state if it's just a challenge.
|
||||
if s["type"] == "webauthn":
|
||||
try:
|
||||
# Get the credential and only include it (not challenges) in the state.
|
||||
s["secret"] = json.loads(s["secret"])["cred_pub_key"]
|
||||
except:
|
||||
# Skip this one --- there is no cred_pub_key.
|
||||
continue
|
||||
mfa_state.append({ "id": s["id"], "type": s["type"], "secret": s["secret"] })
|
||||
return mfa_state
|
||||
|
||||
def enable_mfa(email, type, secret, token, label, env):
|
||||
def enable_mfa(email, type, env, *args):
|
||||
if type == "totp":
|
||||
secret, token, label = args
|
||||
validate_totp_secret(secret)
|
||||
# Sanity check with the provide current token.
|
||||
totp = pyotp.TOTP(secret)
|
||||
if not totp.verify(token, valid_window=1):
|
||||
raise ValueError("Invalid token.")
|
||||
conn, c = open_database(env, with_connection=True)
|
||||
c.execute('INSERT INTO mfa (user_id, type, secret, label) VALUES (?, ?, ?, ?)', (get_user_id(email, c), type, secret, label))
|
||||
conn.commit()
|
||||
elif type == "webauthn":
|
||||
attestationObject, clientDataJSON = args
|
||||
rp = pywarp.RelyingPartyManager(
|
||||
get_relying_party_name(env),
|
||||
rp_id=env["PRIMARY_HOSTNAME"], # must match hostname the control panel is served from
|
||||
credential_storage_backend=WebauthnStorageBackend(env))
|
||||
rp.register(attestation_object=base64.b64decode(attestationObject), client_data_json=base64.b64decode(clientDataJSON), email=email.encode("utf8")) # encoding of email is a little funky here, pywarp calls .decode() with no args?
|
||||
else:
|
||||
raise ValueError("Invalid MFA type.")
|
||||
|
||||
conn, c = open_database(env, with_connection=True)
|
||||
c.execute('INSERT INTO mfa (user_id, type, secret, label) VALUES (?, ?, ?, ?)', (get_user_id(email, c), type, secret, label))
|
||||
conn.commit()
|
||||
|
||||
def set_mru_token(email, mfa_id, token, env):
|
||||
conn, c = open_database(env, with_connection=True)
|
||||
|
|
@ -71,6 +93,9 @@ def validate_totp_secret(secret):
|
|||
if len(secret) != 32:
|
||||
raise ValueError("Secret should be a 32 characters base32 string")
|
||||
|
||||
def get_relying_party_name(env):
|
||||
return env["PRIMARY_HOSTNAME"] + " Mail-in-a-Box Control Panel"
|
||||
|
||||
def provision_totp(email, env):
|
||||
# Make a new secret.
|
||||
secret = base64.b32encode(os.urandom(20)).decode('utf-8')
|
||||
|
|
@ -79,7 +104,7 @@ def provision_totp(email, env):
|
|||
# Make a URI that we encode within a QR code.
|
||||
uri = pyotp.TOTP(secret).provisioning_uri(
|
||||
name=email,
|
||||
issuer_name=env["PRIMARY_HOSTNAME"] + " Mail-in-a-Box Control Panel"
|
||||
issuer_name=get_relying_party_name(env)
|
||||
)
|
||||
|
||||
# Generate a QR code as a base64-encode PNG image.
|
||||
|
|
@ -94,6 +119,55 @@ def provision_totp(email, env):
|
|||
"qr_code_base64": png_b64
|
||||
}
|
||||
|
||||
class WebauthnStorageBackend(pywarp.backends.CredentialStorageBackend):
|
||||
def __init__(self, env):
|
||||
self.env = env
|
||||
def get_record(self, email, conn=None, c=None):
|
||||
# Get an existing record and parse the 'secret' column as JSON.
|
||||
if conn is None: conn, c = open_database(self.env, with_connection=True)
|
||||
c.execute('SELECT secret FROM mfa WHERE user_id=? AND type="webauthn"', (get_user_id(email, c),))
|
||||
config = c.fetchone()
|
||||
if config:
|
||||
try:
|
||||
return json.loads(config[0])
|
||||
except:
|
||||
pass
|
||||
return { }
|
||||
def update_record(self, email, fields):
|
||||
# Update the webauthn record in the database for this user by
|
||||
# merging the fields with the existing fields in the database.
|
||||
conn, c = open_database(self.env, with_connection=True)
|
||||
config = self.get_record(email, conn=conn, c=c)
|
||||
if config:
|
||||
# Merge and update.
|
||||
config.update(fields)
|
||||
config = json.dumps(config)
|
||||
c.execute('UPDATE mfa SET secret=? WHERE user_id=? AND type="webauthn"', (config, get_user_id(email, c),))
|
||||
conn.commit()
|
||||
return
|
||||
|
||||
# Either there's no existing webauthn record or it's corrupted. Delete any existing record.
|
||||
# Then add a new record.
|
||||
c.execute('DELETE FROM mfa WHERE user_id=? AND type="webauthn"', (get_user_id(email, c),))
|
||||
c.execute('INSERT INTO mfa (user_id, type, secret, label) VALUES (?, ?, ?, ?)', (
|
||||
get_user_id(email, c), "webauthn",
|
||||
json.dumps(fields),
|
||||
"WebAuthn"))
|
||||
conn.commit()
|
||||
def save_challenge_for_user(self, email, challenge, type):
|
||||
self.update_record(email, { type + "challenge": base64.b64encode(challenge).decode("ascii") })
|
||||
def get_challenge_for_user(self, email, type):
|
||||
challenge = self.get_record(email).get(type + "challenge")
|
||||
if challenge: challenge = base64.b64decode(challenge.encode("ascii"))
|
||||
return challenge
|
||||
|
||||
def provision_webauthn(email, env):
|
||||
rp = pywarp.RelyingPartyManager(
|
||||
get_relying_party_name(env),
|
||||
rp_id=env["PRIMARY_HOSTNAME"], # must match hostname the control panel is served from
|
||||
credential_storage_backend=WebauthnStorageBackend(env))
|
||||
return rp.get_registration_options(email=email)
|
||||
|
||||
def validate_auth_mfa(email, request, env):
|
||||
# Validates that a login request satisfies any MFA modes
|
||||
# that have been enabled for the user's account. Returns
|
||||
|
|
|
|||
|
|
@ -62,6 +62,14 @@ and ensure every administrator account for this control panel does the same.</st
|
|||
</div>
|
||||
</form>
|
||||
|
||||
<form id="webauthn-setup" style="display: none">
|
||||
<h3>Add a WebAuthn Device</h3>
|
||||
<p>If you have a WebAuthn device such as a YubiKey, plug it in and click Add WebAuthn Device.</p>
|
||||
<button type="submit" class="btn" onclick="return do_enable_webauthn()">Add WebAuthn Device</button>
|
||||
</form>
|
||||
|
||||
<div id="webauthn-setup" style="display: none">
|
||||
</div>
|
||||
|
||||
<div id="output-2fa" class="panel panel-danger hidden">
|
||||
<div class="panel-body"></div>
|
||||
|
|
@ -87,6 +95,7 @@ and ensure every administrator account for this control panel does the same.</st
|
|||
totpSetupLabel: document.getElementById('totp-setup-label'),
|
||||
totpQr: document.getElementById('totp-setup-qr'),
|
||||
totpSetupSubmit: document.querySelector('#totp-setup-submit'),
|
||||
webauthnSetupForm: document.getElementById('webauthn-setup'),
|
||||
wrapper: document.querySelector('.twofactor')
|
||||
}
|
||||
|
||||
|
|
@ -126,6 +135,16 @@ and ensure every administrator account for this control panel does the same.</st
|
|||
el.wrapper.classList.add('disabled');
|
||||
}
|
||||
|
||||
function arrayBufferToBase64(a) { return btoa(String.fromCharCode(...new Uint8Array(a))); }
|
||||
function base64ToArrayBuffer(b) { return Uint8Array.from(atob(b), c => c.charCodeAt(0)); }
|
||||
|
||||
function render_webauthn_setup(provisioning) {
|
||||
$(el.webauthnSetupForm).show();
|
||||
provisioning.challenge = base64ToArrayBuffer(provisioning.challenge);
|
||||
provisioning.user.id = new TextEncoder().encode(provisioning.user.name);
|
||||
window.mailinabix_mfa_webauthn_provision = provisioning;
|
||||
}
|
||||
|
||||
function render_disable(mfa) {
|
||||
var panel = $('#mfa-device-templates .' + mfa.type).clone();
|
||||
$('#mfa-devices').append(panel);
|
||||
|
|
@ -158,6 +177,8 @@ and ensure every administrator account for this control panel does the same.</st
|
|||
el.totpSetupToken.removeEventListener('input', update_setup_disabled);
|
||||
el.totpSetupSubmit.setAttribute('disabled', '');
|
||||
el.totpQr.innerHTML = '';
|
||||
|
||||
$(el.webauthnSetupForm).hide();
|
||||
}
|
||||
|
||||
function show_mfa() {
|
||||
|
|
@ -178,6 +199,8 @@ and ensure every administrator account for this control panel does the same.</st
|
|||
|
||||
if (res.new_mfa.totp)
|
||||
render_totp_setup(res.new_mfa.totp);
|
||||
if (res.new_mfa.webauthn && 'credentials' in navigator)
|
||||
render_webauthn_setup(res.new_mfa.webauthn);
|
||||
}
|
||||
);
|
||||
}
|
||||
|
|
@ -216,4 +239,22 @@ and ensure every administrator account for this control panel does the same.</st
|
|||
|
||||
return false;
|
||||
}
|
||||
|
||||
function do_enable_webauthn() {
|
||||
navigator.credentials.create({ publicKey: window.mailinabix_mfa_webauthn_provision })
|
||||
.then(function(creds) {
|
||||
api(
|
||||
'/mfa/enable/webauthn',
|
||||
'POST',
|
||||
{
|
||||
attestationObject: arrayBufferToBase64(creds.response['attestationObject']),
|
||||
clientDataJSON: arrayBufferToBase64(creds.response['clientDataJSON'])
|
||||
},
|
||||
function(res) { do_logout(); },
|
||||
function(res) { render_error(res); }
|
||||
);
|
||||
});
|
||||
return false;
|
||||
}
|
||||
|
||||
</script>
|
||||
|
|
|
|||
|
|
@ -50,7 +50,7 @@ hide_output $venv/bin/pip install --upgrade pip
|
|||
hide_output $venv/bin/pip install --upgrade \
|
||||
rtyaml "email_validator>=1.0.0" "exclusiveprocess" \
|
||||
flask dnspython python-dateutil expiringdict \
|
||||
qrcode[pil] pyotp \
|
||||
qrcode[pil] pyotp pywarp \
|
||||
"idna>=2.0.0" "cryptography==2.2.2" boto psutil postfix-mta-sts-resolver b2sdk
|
||||
|
||||
# CONFIGURATION
|
||||
|
|
|
|||
Loading…
Reference in New Issue