feat: 支持 passkey 登录 (#11519)

* perf: 基本完成功能

* perf: 优化 passkey

* perf: 优化 passkey

* perf: 完成 passkey

---------

Co-authored-by: ibuler <ibuler@qq.com>
This commit is contained in:
fit2bot
2023-09-11 18:15:03 +08:00
committed by GitHub
parent d7ca1a09d4
commit 72b215ed03
37 changed files with 899 additions and 144 deletions

View File

@@ -0,0 +1 @@
from .backends import *

View File

@@ -0,0 +1,59 @@
from django.conf import settings
from django.http import JsonResponse
from django.shortcuts import render
from django.utils.translation import gettext as _
from rest_framework.decorators import action
from rest_framework.permissions import IsAuthenticated, AllowAny
from rest_framework.viewsets import ModelViewSet
from authentication.mixins import AuthMixin
from .fido import register_begin, register_complete, auth_begin, auth_complete
from .models import Passkey
from .serializer import PasskeySerializer
from ...views import FlashMessageMixin
class PasskeyViewSet(AuthMixin, FlashMessageMixin, ModelViewSet):
serializer_class = PasskeySerializer
permission_classes = (IsAuthenticated,)
def get_queryset(self):
return Passkey.objects.filter(user=self.request.user)
@action(methods=['get', 'post'], detail=False, url_path='register')
def register(self, request):
if request.method == 'GET':
register_data, state = register_begin(request)
return JsonResponse(dict(register_data))
else:
passkey = register_complete(request)
return JsonResponse({'id': passkey.id.__str__(), 'name': passkey.name})
@action(methods=['get'], detail=False, url_path='login', permission_classes=[AllowAny])
def login(self, request):
return render(request, 'authentication/passkey.html', {})
def redirect_to_error(self, error):
self.send_auth_signal(success=False, username='unknown', reason='passkey')
return render(self.request, 'authentication/passkey.html', {'error': error})
@action(methods=['get', 'post'], detail=False, url_path='auth', permission_classes=[AllowAny])
def auth(self, request):
if request.method == 'GET':
auth_data = auth_begin(request)
return JsonResponse(dict(auth_data))
try:
user = auth_complete(request)
except ValueError as e:
return self.redirect_to_error(str(e))
if not user:
return self.redirect_to_error(_('Auth failed'))
try:
self.check_oauth2_auth(user, settings.AUTH_BACKEND_PASSKEY)
return self.redirect_to_guard_view()
except Exception as e:
msg = getattr(e, 'msg', '') or str(e)
return self.redirect_to_error(msg)

View File

@@ -0,0 +1,9 @@
from django.conf import settings
from ..base import JMSModelBackend
class PasskeyAuthBackend(JMSModelBackend):
@staticmethod
def is_enabled():
return settings.AUTH_PASSKEY

View File

@@ -0,0 +1,155 @@
import json
from urllib.parse import urlparse
import fido2.features
from django.conf import settings
from django.utils import timezone
from django.utils.translation import gettext as _
from fido2.server import Fido2Server
from fido2.utils import websafe_decode, websafe_encode
from fido2.webauthn import PublicKeyCredentialRpEntity, AttestedCredentialData, PublicKeyCredentialUserEntity
from rest_framework.serializers import ValidationError
from user_agents.parsers import parse as ua_parse
from common.utils import get_logger
from .models import Passkey
logger = get_logger(__name__)
try:
fido2.features.webauthn_json_mapping.enabled = True
except:
pass
def get_current_platform(request):
ua = ua_parse(request.META["HTTP_USER_AGENT"])
if 'Safari' in ua.browser.family:
return "Apple"
elif 'Chrome' in ua.browser.family and ua.os.family == "Mac OS X":
return "Chrome on Apple"
elif 'Android' in ua.os.family:
return "Google"
elif "Windows" in ua.os.family:
return "Microsoft"
else:
return "Key"
def get_server_id_from_request(request, allowed=()):
origin = request.META.get('HTTP_REFERER')
if not origin:
origin = request.get_host()
p = urlparse(origin)
if p.netloc in allowed or p.hostname in allowed:
return p.hostname
else:
return 'localhost'
def default_server_id(request):
domains = settings.ALLOWED_DOMAINS
return get_server_id_from_request(request, allowed=domains)
def get_server(request=None):
"""Get Server Info from settings and returns a Fido2Server"""
server_id = settings.FIDO_SERVER_ID or default_server_id(request)
if callable(server_id):
fido_server_id = settings.FIDO_SERVER_ID(request)
elif ',' in server_id:
fido_server_id = get_server_id_from_request(request, allowed=server_id.split(','))
else:
fido_server_id = server_id
logger.debug('Fido server id: {}'.format(fido_server_id))
if callable(settings.FIDO_SERVER_NAME):
fido_server_name = settings.FIDO_SERVER_NAME(request)
else:
fido_server_name = settings.FIDO_SERVER_NAME
rp = PublicKeyCredentialRpEntity(id=fido_server_id, name=fido_server_name)
return Fido2Server(rp)
def get_user_credentials(username):
user_passkeys = Passkey.objects.filter(user__username=username)
return [AttestedCredentialData(websafe_decode(uk.token)) for uk in user_passkeys]
def register_begin(request):
server = get_server(request)
user = request.user
user_credentials = get_user_credentials(user.username)
prefix = request.query_params.get('name', '')
prefix = '(' + prefix + ')'
user_entity = PublicKeyCredentialUserEntity(
id=str(user.id).encode('utf8'),
name=user.username + prefix,
display_name=user.name,
)
auth_attachment = getattr(settings, 'KEY_ATTACHMENT', None)
data, state = server.register_begin(
user_entity, user_credentials,
authenticator_attachment=auth_attachment,
resident_key_requirement=fido2.webauthn.ResidentKeyRequirement.PREFERRED
)
request.session['fido2_state'] = state
data = dict(data)
return data, state
def register_complete(request):
if not request.session.get("fido2_state"):
raise ValidationError("No state found")
data = request.data
server = get_server(request)
state = request.session.pop("fido2_state")
auth_data = server.register_complete(state, response=data)
encoded = websafe_encode(auth_data.credential_data)
platform = get_current_platform(request)
name = data.pop("key_name", '') or platform
passkey = Passkey.objects.create(
user=request.user,
token=encoded,
name=name,
platform=platform,
credential_id=data.get('id')
)
return passkey
def auth_begin(request):
server = get_server(request)
credentials = []
username = None
if request.user.is_authenticated:
username = request.user.username
if username:
credentials = get_user_credentials(username)
auth_data, state = server.authenticate_begin(credentials)
request.session['fido2_state'] = state
return auth_data
def auth_complete(request):
server = get_server(request)
data = request.data.get("passkeys")
data = json.loads(data)
cid = data['id']
key = Passkey.objects.filter(credential_id=cid, is_active=True).first()
if not key:
raise ValueError(_("This key is not registered"))
credentials = [AttestedCredentialData(websafe_decode(key.token))]
state = request.session.get('fido2_state')
server.authenticate_complete(state, credentials=credentials, response=data)
request.session["passkey"] = '{}_{}'.format(key.id, key.name)
key.date_last_used = timezone.now()
key.save(update_fields=['date_last_used'])
return key.user

View File

@@ -0,0 +1,19 @@
from django.conf import settings
from django.db import models
from django.utils.translation import gettext_lazy as _
from common.db.models import JMSBaseModel
class Passkey(JMSBaseModel):
user = models.ForeignKey(settings.AUTH_USER_MODEL, on_delete=models.CASCADE)
name = models.CharField(max_length=255, verbose_name=_("Name"))
is_active = models.BooleanField(default=True, verbose_name=_("Enabled"))
platform = models.CharField(max_length=255, default='', verbose_name=_("Platform"))
added_on = models.DateTimeField(auto_now_add=True, verbose_name=_("Added on"))
date_last_used = models.DateTimeField(null=True, default=None, verbose_name=_("Date last used"))
credential_id = models.CharField(max_length=255, unique=True, null=False, verbose_name=_("Credential ID"))
token = models.CharField(max_length=255, null=False, verbose_name=_("Token"))
def __str__(self):
return self.name

View File

@@ -0,0 +1,13 @@
from rest_framework import serializers
from .models import Passkey
class PasskeySerializer(serializers.ModelSerializer):
class Meta:
model = Passkey
fields = [
'id', 'name', 'is_active', 'created_by',
'date_last_used', 'date_created',
]
read_only_fields = list(set(fields) - {'is_active'})

View File

@@ -0,0 +1,9 @@
from rest_framework.routers import DefaultRouter
from . import api
router = DefaultRouter()
router.register('passkeys', api.PasskeyViewSet, 'passkey')
urlpatterns = []
urlpatterns += router.urls