Added Webautn login

TODO: modify the middleware to certify the user has successfully used it...
This commit is contained in:
Spitfireap
2024-06-22 19:45:23 +02:00
parent dccd22bdaa
commit 381e4c7be5
9 changed files with 429 additions and 47 deletions
+32 -11
View File
@@ -75,7 +75,8 @@ class CoreGlobalParametersSerializer(serializers.Serializer):
default_password = serializers.CharField(default="ChangeMe1!")
random_password_length = serializers.IntegerField(min_value=8, default=8)
update_password_url = serializers.URLField(required=False, allow_blank=True)
password_recovery_msg = serializers.CharField(required=False, allow_blank=True)
password_recovery_msg = serializers.CharField(
required=False, allow_blank=True)
sms_password_recovery = serializers.BooleanField(default=False)
sms_provider = serializers.ChoiceField(
choices=constants.SMS_BACKENDS, required=False, allow_null=True
@@ -86,7 +87,8 @@ class CoreGlobalParametersSerializer(serializers.Serializer):
ldap_server_port = serializers.IntegerField(default=389)
ldap_enable_secondary_server = serializers.BooleanField(default=False)
ldap_secondary_server_address = serializers.CharField(required=False)
ldap_secondary_server_port = serializers.IntegerField(default=389, required=False)
ldap_secondary_server_port = serializers.IntegerField(
default=389, required=False)
ldap_secured = serializers.ChoiceField(
choices=constants.LDAP_SECURE_MODES, default="none"
)
@@ -107,7 +109,8 @@ class CoreGlobalParametersSerializer(serializers.Serializer):
choices=constants.LDAP_AUTH_METHODS,
default="searchbind",
)
ldap_bind_dn = serializers.CharField(default="", required=False, allow_blank=True)
ldap_bind_dn = serializers.CharField(
default="", required=False, allow_blank=True)
ldap_bind_password = serializers.CharField(
default="", required=False, allow_blank=True
)
@@ -123,15 +126,18 @@ class CoreGlobalParametersSerializer(serializers.Serializer):
# LDAP sync settings
ldap_sync_bind_dn = serializers.CharField(required=False, allow_blank=True)
ldap_sync_bind_password = serializers.CharField(required=False, allow_blank=True)
ldap_sync_bind_password = serializers.CharField(
required=False, allow_blank=True)
ldap_enable_sync = serializers.BooleanField(default=False)
ldap_sync_delete_remote_account = serializers.BooleanField(default=False)
ldap_sync_account_dn_template = serializers.CharField(
required=False, allow_blank=True
)
ldap_enable_import = serializers.BooleanField(default=False)
ldap_import_search_base = serializers.CharField(required=False, allow_blank=True)
ldap_import_search_filter = serializers.CharField(default="(cn=*)", required=False)
ldap_import_search_base = serializers.CharField(
required=False, allow_blank=True)
ldap_import_search_filter = serializers.CharField(
default="(cn=*)", required=False)
ldap_import_username_attr = serializers.CharField(default="cn")
ldap_dovecot_sync = serializers.BooleanField(default=False)
ldap_dovecot_conf_file = serializers.CharField(
@@ -145,7 +151,8 @@ class CoreGlobalParametersSerializer(serializers.Serializer):
hide_features_widget = serializers.BooleanField(default=False)
# Notification settings
sender_address = lib_fields.DRFEmailFieldUTF8(default="noreply@yourdomain.test")
sender_address = lib_fields.DRFEmailFieldUTF8(
default="noreply@yourdomain.test")
# API settings
enable_api_communication = serializers.BooleanField(default=True)
@@ -255,6 +262,16 @@ class FidoRegistrationSerializer(serializers.Serializer):
name = serializers.CharField()
class FidoAuthenticationSerializer(serializers.Serializer):
"""Serializer used to finish the fido key authentication."""
authenticatorAttachment = serializers.CharField()
clientExtensionResults = serializers.JSONField()
id = serializers.CharField()
rawId = serializers.CharField()
response = serializers.JSONField()
type = serializers.CharField()
class LogSerializer(serializers.ModelSerializer):
"""Log serializer."""
@@ -363,7 +380,8 @@ class PasswordRecoverySmsSerializer(serializers.Serializer):
request = self.context["request"]
clean_email = data["email"]
user = User.objects.filter(email__iexact=clean_email, is_active=True).first()
user = User.objects.filter(
email__iexact=clean_email, is_active=True).first()
if user is None:
raise CustomValidationError(
{"type": "sms", "reason": "No valid user found."}, 404
@@ -383,7 +401,8 @@ class PasswordRecoverySmsSerializer(serializers.Serializer):
user = self.context["user"]
if not user:
raise NoSMSAvailable()
backend = sms_backends.get_active_backend(request.localconfig.parameters)
backend = sms_backends.get_active_backend(
request.localconfig.parameters)
secret = cryptutils.random_hex_key(20)
send_sms_code(secret, backend, user)
request.session["totp_secret"] = secret
@@ -446,7 +465,8 @@ class PasswordRecoverySmsResendSerializer(serializers.Serializer):
backend = sms_backends.get_active_backend(
self.context["request"].localconfig.parameters
)
send_sms_code(self.context["totp_secret"], backend, self.context["user"])
send_sms_code(self.context["totp_secret"],
backend, self.context["user"])
return True
@@ -475,7 +495,8 @@ class PasswordRecoveryConfirmSerializer(serializers.Serializer):
data["new_password1"] == ""
or data["new_password1"] != data["new_password2"]
):
raise serializers.ValidationError("Password is empty or does not match")
raise serializers.ValidationError(
"Password is empty or does not match")
user = self.get_user(data["id"])
+23 -3
View File
@@ -64,7 +64,8 @@ class AccountViewSet(core_v1_viewsets.AccountViewSet):
def manage_api_token(self, request):
"""Manage API token."""
if not request.user.is_superuser:
raise PermissionDenied("Only super administrators can have API tokens")
raise PermissionDenied(
"Only super administrators can have API tokens")
if request.method == "DELETE":
Token.objects.filter(user=request.user).delete()
return response.Response(status=204)
@@ -196,9 +197,28 @@ class FIDOViewSet(GetThrottleViewsetMixin,
data=request.data)
serializer.is_valid(raise_exception=True)
credential_data = f2_auth.end_registration(request.data,
request.session.pop("fido2_state"),
request.user.id)
request.session.pop(
"fido2_state"),
request.user.id)
models.UserFidoKeys.objects.create(name=request.data["name"],
credential_data=credential_data,
user=request.user)
if not request.user.tfa_enabled:
request.user.tfa_enabled = True
request.user.save()
return response.Response({"success": True}, 200)
@action(methods=["post"], detail=False, url_path="authenticate/begin")
def authenticate_begin(self, request):
"""An API View to start the authentication process."""
options = f2_auth.begin_authentication(request)
return response.Response(dict(options))
@action(methods=["post"], detail=False, url_path="authenticate/end")
def authenticate_end(self, request):
"""An API View to finish the authentication process."""
serializer = serializers.FidoAuthenticationSerializer(
data=request.data)
serializer.is_valid(raise_exception=True)
result, detail = f2_auth.end_authentication(request)
return response.Response({"success": True})
+30 -2
View File
@@ -5,6 +5,7 @@ from fido2.webauthn import PublicKeyCredentialRpEntity, PublicKeyCredentialUserE
from fido2.server import Fido2Server
from fido2.utils import websafe_decode, websafe_encode
import fido2.features
import datetime
def set_json_mapping():
@@ -34,7 +35,7 @@ def begin_registration(request):
),
get_creds_from_user(request.user.pk),
user_verification=UserVerificationRequirement.DISCOURAGED,
extensions={"credentialProtectionPolicy":"userVerificationOptional"}
extensions={"credentialProtectionPolicy": "userVerificationOptional"}
)
request.session['fido2_state'] = state
return options
@@ -46,5 +47,32 @@ def end_registration(data, fido2_state, user_id):
auth_data = server.register_complete(
fido2_state,
data
)
)
return websafe_encode(auth_data.credential_data)
def begin_authentication(request):
set_json_mapping()
server = create_fido2_server()
options, state = server.authenticate_begin(
get_creds_from_user(request.user.id))
request.session['fido_state'] = state
return options
def end_authentication(request):
set_json_mapping()
server = create_fido2_server()
credentials = get_creds_from_user(request.user.id)
result = server.authenticate_complete(
request.session.pop("fido_state"),
credentials,
request.data,
)
for key in UserFidoKeys.objects.filter(user=request.user.id):
cred = AttestedCredentialData(websafe_decode(key.credential_data))
if cred.credential_id == result.credential_id:
key.last_used = datetime.datetime.now()
key.use_count += 1
key.save()
return True, ""
+17 -1
View File
@@ -9,7 +9,8 @@ from django.utils.translation import gettext as _, gettext_lazy
import django_otp
from modoboa.core.models import User
from modoboa.core.models import User, UserFidoKeys
from modoboa.core import utils
from modoboa.lib.form_utils import UserKwargModelFormMixin
from modoboa.parameters import tools as param_tools
@@ -117,6 +118,21 @@ class ProfileForm(forms.ModelForm):
return user
class TwoFAChoiceForm(forms.Form):
"""Form to select the 2FA method of choice."""
two_fa_choices = forms.ChoiceField(choices=(),
required=True)
def __init__(self, *args, **kwargs):
user = kwargs.pop("user")
fido_keys = UserFidoKeys.objects.filter(user=user)
if fido_keys.exists():
self.fields["two_fa_choices.choices"].choices = [
("TOTP", "TOTP or recovery codes"),
("FIDO", "Webauthn device")
]
class APIAccessForm(forms.Form):
"""Form to control API access."""
+2 -1
View File
@@ -16,7 +16,8 @@ class LocalConfigMiddleware(MiddlewareMixin):
class TwoFAMiddleware:
"""Custom 2FA middleware to enforce verification if user has TFA enabled."""
"""Custom 2FA middleware to enforce verification if user has TFA enabled.
Also verify the user if it successfully use a webautn token."""
def __init__(self, get_response=None):
self.get_response = get_response
+11 -6
View File
@@ -58,7 +58,8 @@ class User(AbstractUser):
master_user = models.BooleanField(
gettext_lazy("Allow mailboxes access"),
default=False,
help_text=gettext_lazy("Allow this administrator to access user mailboxes"),
help_text=gettext_lazy(
"Allow this administrator to access user mailboxes"),
)
password = models.CharField(gettext_lazy("password"), max_length=256)
@@ -69,7 +70,8 @@ class User(AbstractUser):
choices=constants.LANGUAGES,
help_text=gettext_lazy("Prefered language to display pages."),
)
phone_number = PhoneNumberField(gettext_lazy("Phone number"), blank=True, null=True)
phone_number = PhoneNumberField(gettext_lazy(
"Phone number"), blank=True, null=True)
secondary_email = models.EmailField(
gettext_lazy("Secondary email"),
max_length=254,
@@ -212,7 +214,8 @@ class User(AbstractUser):
"""
ct = ContentType.objects.get_for_model(obj)
try:
ooentry = self.objectaccess_set.get(content_type=ct, object_id=obj.id)
ooentry = self.objectaccess_set.get(
content_type=ct, object_id=obj.id)
except ObjectAccess.DoesNotExist:
return False
return ooentry.is_owner
@@ -233,7 +236,8 @@ class User(AbstractUser):
ct = ContentType.objects.get_for_model(obj)
try:
ooentry = self.objectaccess_set.get(content_type=ct, object_id=obj.id)
ooentry = self.objectaccess_set.get(
content_type=ct, object_id=obj.id)
except ObjectAccess.DoesNotExist:
pass
else:
@@ -394,7 +398,8 @@ class User(AbstractUser):
smart_str(self.role),
smart_str(self.email),
]
results = signals.account_exported.send(sender=self.__class__, user=self)
results = signals.account_exported.send(
sender=self.__class__, user=self)
for result in results:
row += result[1]
return row
@@ -413,7 +418,7 @@ reversion.register(User)
class UserFidoKeys(models.Model):
user = models.ForeignKey(User ,on_delete=models.CASCADE)
user = models.ForeignKey(User, on_delete=models.CASCADE)
name = models.CharField(max_length=255)
enabled = models.BooleanField(default=True)
added_on = models.DateTimeField(auto_now_add=True)
+23 -7
View File
@@ -86,13 +86,15 @@ def dologin(request):
translation.activate(request.user.language)
logger.info(_("User '%s' successfully logged in") % user.username)
logger.info(_("User '%s' successfully logged in") %
user.username)
signals.user_login.send(
sender="dologin",
username=form.cleaned_data["username"],
password=form.cleaned_data["password"],
)
response = HttpResponseRedirect(find_nextlocation(request, user))
response = HttpResponseRedirect(
find_nextlocation(request, user))
response.set_cookie(
settings.LANGUAGE_COOKIE_NAME, request.user.language
)
@@ -114,7 +116,8 @@ def dologin(request):
nextlocation = request.GET.get("next", "")
httpcode = 200
announcements = signals.get_announcements.send(sender="login", location="loginpage")
announcements = signals.get_announcements.send(
sender="login", location="loginpage")
announcements = [announcement[1] for announcement in announcements]
return HttpResponse(
render_to_string(
@@ -154,7 +157,8 @@ class PasswordResetView(auth_views.PasswordResetView):
def setup(self, request, *args, **kwargs):
super().setup(request, *args, **kwargs)
self.from_email = request.localconfig.parameters.get_value("sender_address")
self.from_email = request.localconfig.parameters.get_value(
"sender_address")
def get_context_data(self, **kwargs):
"""Include help text."""
@@ -177,7 +181,8 @@ class PasswordResetView(auth_views.PasswordResetView):
if not user:
# Fallback to email
return super().form_valid(form)
backend = sms_backends.get_active_backend(self.request.localconfig.parameters)
backend = sms_backends.get_active_backend(
self.request.localconfig.parameters)
secret = cryptutils.random_hex_key(20)
code = oath.totp(secret)
text = _(
@@ -227,10 +232,12 @@ class ResendSMSCodeView(generic.View):
if not sms_password_recovery:
raise Http404
try:
user = models.User._default_manager.get(pk=self.request.session["user_pk"])
user = models.User._default_manager.get(
pk=self.request.session["user_pk"])
except KeyError:
raise Http404
backend = sms_backends.get_active_backend(self.request.localconfig.parameters)
backend = sms_backends.get_active_backend(
self.request.localconfig.parameters)
secret = cryptutils.random_hex_key(20)
code = oath.totp(secret)
text = _(
@@ -252,6 +259,15 @@ class TwoFactorCodeVerifyView(
form_class = forms.Verify2FACodeForm
template_name = "registration/twofactor_code_verify.html"
def get_context_data(self, **kwargs):
context = super().get_context_data(**kwargs)
has_totp_device = django_otp.user_has_device(self.request.user)
context["totp_device"] = has_totp_device
has_webauthn_device = models.UserFidoKeys.objects.filter(
user=self.request.user).exists()
context["webauthn_device"] = has_webauthn_device
return context
def form_valid(self, form):
"""Login user."""
django_otp.login(self.request, form.cleaned_data["tfa_code"])
@@ -0,0 +1,193 @@
// src/webauthn-json/base64url.ts
function base64urlToBuffer(baseurl64String) {
const padding = "==".slice(0, (4 - baseurl64String.length % 4) % 4);
const base64String = baseurl64String.replace(/-/g, "+").replace(/_/g, "/") + padding;
const str = atob(base64String);
const buffer = new ArrayBuffer(str.length);
const byteView = new Uint8Array(buffer);
for (let i = 0; i < str.length; i++) {
byteView[i] = str.charCodeAt(i);
}
return buffer;
}
function bufferToBase64url(buffer) {
const byteView = new Uint8Array(buffer);
let str = "";
for (const charCode of byteView) {
str += String.fromCharCode(charCode);
}
const base64String = btoa(str);
const base64urlString = base64String.replace(/\+/g, "-").replace(/\//g, "_").replace(/=/g, "");
return base64urlString;
}
// src/webauthn-json/convert.ts
var copyValue = "copy";
var convertValue = "convert";
function convert(conversionFn, schema, input) {
if (schema === copyValue) {
return input;
}
if (schema === convertValue) {
return conversionFn(input);
}
if (schema instanceof Array) {
return input.map((v) => convert(conversionFn, schema[0], v));
}
if (schema instanceof Object) {
const output = {};
for (const [key, schemaField] of Object.entries(schema)) {
if (schemaField.derive) {
const v = schemaField.derive(input);
if (v !== void 0) {
input[key] = v;
}
}
if (!(key in input)) {
if (schemaField.required) {
throw new Error(`Missing key: ${key}`);
}
continue;
}
if (input[key] == null) {
output[key] = null;
continue;
}
output[key] = convert(conversionFn, schemaField.schema, input[key]);
}
return output;
}
}
function derived(schema, derive) {
return {
required: true,
schema,
derive
};
}
function required(schema) {
return {
required: true,
schema
};
}
function optional(schema) {
return {
required: false,
schema
};
}
// src/webauthn-json/basic/schema.ts
var publicKeyCredentialDescriptorSchema = {
type: required(copyValue),
id: required(convertValue),
transports: optional(copyValue)
};
var simplifiedExtensionsSchema = {
appid: optional(copyValue),
appidExclude: optional(copyValue),
credProps: optional(copyValue)
};
var simplifiedClientExtensionResultsSchema = {
appid: optional(copyValue),
appidExclude: optional(copyValue),
credProps: optional(copyValue)
};
var credentialCreationOptions = {
publicKey: required({
rp: required(copyValue),
user: required({
id: required(convertValue),
name: required(copyValue),
displayName: required(copyValue)
}),
challenge: required(convertValue),
pubKeyCredParams: required(copyValue),
timeout: optional(copyValue),
excludeCredentials: optional([publicKeyCredentialDescriptorSchema]),
authenticatorSelection: optional(copyValue),
attestation: optional(copyValue),
extensions: optional(simplifiedExtensionsSchema)
}),
signal: optional(copyValue)
};
var publicKeyCredentialWithAttestation = {
type: required(copyValue),
id: required(copyValue),
rawId: required(convertValue),
authenticatorAttachment: optional(copyValue),
response: required({
clientDataJSON: required(convertValue),
attestationObject: required(convertValue),
transports: derived(copyValue, (response) => {
var _a;
return ((_a = response.getTransports) == null ? void 0 : _a.call(response)) || [];
})
}),
clientExtensionResults: derived(simplifiedClientExtensionResultsSchema, (pkc) => pkc.getClientExtensionResults())
};
var credentialRequestOptions = {
mediation: optional(copyValue),
publicKey: required({
challenge: required(convertValue),
timeout: optional(copyValue),
rpId: optional(copyValue),
allowCredentials: optional([publicKeyCredentialDescriptorSchema]),
userVerification: optional(copyValue),
extensions: optional(simplifiedExtensionsSchema)
}),
signal: optional(copyValue)
};
var publicKeyCredentialWithAssertion = {
type: required(copyValue),
id: required(copyValue),
rawId: required(convertValue),
authenticatorAttachment: optional(copyValue),
response: required({
clientDataJSON: required(convertValue),
authenticatorData: required(convertValue),
signature: required(convertValue),
userHandle: required(convertValue)
}),
clientExtensionResults: derived(simplifiedClientExtensionResultsSchema, (pkc) => pkc.getClientExtensionResults())
};
// src/webauthn-json/basic/api.ts
function createRequestFromJSON(requestJSON) {
return convert(base64urlToBuffer, credentialCreationOptions, requestJSON);
}
function createResponseToJSON(credential) {
return convert(bufferToBase64url, publicKeyCredentialWithAttestation, credential);
}
function getRequestFromJSON(requestJSON) {
return convert(base64urlToBuffer, credentialRequestOptions, requestJSON);
}
function getResponseToJSON(credential) {
return convert(bufferToBase64url, publicKeyCredentialWithAssertion, credential);
}
// src/webauthn-json/basic/supported.ts
function supported() {
return !!(navigator.credentials && navigator.credentials.create && navigator.credentials.get && window.PublicKeyCredential);
}
// src/webauthn-json/browser-ponyfill.ts
async function create(options) {
const response = await navigator.credentials.create(options);
response.toJSON = () => createResponseToJSON(response);
return response;
}
async function get(options) {
const response = await navigator.credentials.get(options);
response.toJSON = () => getResponseToJSON(response);
return response;
}
export {
create,
get,
createRequestFromJSON as parseCreationOptionsFromJSON,
getRequestFromJSON as parseRequestOptionsFromJSON,
supported
};
//# sourceMappingURL=webauthn-json.browser-ponyfill.js.map
@@ -9,22 +9,104 @@
{% block main_content %}
<div class="row">
<div class="col-sm-offset-4 col-sm-6">
<form method="post" action="{% url 'core:2fa_verify' %}" class="login-form">
{% csrf_token %}
<div class="form-group{% if form.tfa_code.errors %} has-error{% endif %}">
<label class="control-label">Two-Factor authentication code</label>
<input id="id_tfa_code" name="tfa_code" class="form-control">
<span class="help-block">
{{ form.tfa_code.errors }}
</span>
<p class="help-block">
Enter the code from the two-factor app on your mobile
device. If you've lost your device, you may enter one of your
recovery codes.
</p>
</div>
<button class="btn btn-primary col-sm-12" type="submit">Verify code</button>
</form>
<div id="methodSelection">
{% if totp_device %}
<button onclick="showTOTP()">Use TOTP/backup code</button>
{% endif %}
{% if webauthn_device %}
<button onclick="showWebAuthn()">Use Webautn device</button>
{% endif %}
</div>
<div id="totp_form" style="display:none;">
<form method="post" action="{% url 'core:2fa_verify' %}" class="login-form">
{% csrf_token %}
<div class="form-group{% if form.tfa_code.errors %} has-error{% endif %}">
<label class="control-label">Two-Factor authentication code</label>
<input id="id_tfa_code" name="tfa_code" class="form-control">
<span class="help-block">
{{ form.tfa_code.errors }}
</span>
<p class="help-block">
Enter the code from the two-factor app on your mobile
device. If you've lost your device, you may enter one of your
recovery codes.
</p>
</div>
<button class="btn btn-primary col-sm-12" type="submit">Verify code</button>
</form>
<button onclick="backToSelection()">Back</button>
</div>
<div id="webautn_form" style="display:none;">
<button onclick="useWebAuthn()">Log using Webauthn</button>
<button onclick="backToSelection()">Back</button>
</div>
</div>
</div>
{% endblock %}
{% block extra_js %}
<script type="module" defer>
import {
get,
parseRequestOptionsFromJSON,
} from "{% static 'js/webauthn-json.browser-ponyfill.js' %}"
function showTOTP() {
document.getElementById("totp_form").style.display = 'block'
document.getElementById("methodSelection").style.display = 'none'
}
window.showTOTP = showTOTP
function backToSelection() {
document.getElementById("totp_form").style.display = 'none'
document.getElementById("webautn_form").style.display = 'none'
document.getElementById("methodSelection").style.display = 'block'
}
window.backToSelection = backToSelection
function showWebAuthn() {
document.getElementById("webautn_form").style.display = 'block'
document.getElementById("methodSelection").style.display = 'none'
}
window.showWebAuthn = showWebAuthn
async function useWebAuthn() {
const csrf = getCookie('csrftoken')
console.log(csrf)
const request = await fetch("{% url 'v2:fido-authenticate-begin' %}", {
method: 'POST',
headers: { "X-CSRFToken": csrf, 'Content-Type': 'application/json' }
})
if(!request.ok) {
throw new Error('No credential available to authenticate!')
}
let json = await request.json()
let options = parseRequestOptionsFromJSON(json)
let response = await get(options);
let result = await fetch("{% url 'v2:fido-authenticate-end' %}", {
method: 'POST',
headers: { "X-CSRFToken": csrf, 'Content-Type': 'application/json'},
body: JSON.stringify(response),
})
}
window.useWebAuthn = useWebAuthn
function getCookie(cname) {
const name = cname + "="
const decodedCookie = decodeURIComponent(document.cookie)
const ca = decodedCookie.split(';')
for(let i = 0; i <ca.length; i++) {
let c = ca[i]
while (c.charAt(0) == ' ') {
c = c.substring(1)
}
if (c.indexOf(name) == 0) {
return c.substring(name.length, c.length)
}
}
return ""
}
</script>
{% endblock %}