Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
19 changes: 18 additions & 1 deletion apps/account/admin.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,7 +4,7 @@
from django.utils.translation import gettext_lazy
from ovinc_client.core.utils import strtobool

from apps.account.models import User, UserToken
from apps.account.models import User, UserPasskeyCredential, UserToken


@admin.register(User)
Expand Down Expand Up @@ -43,3 +43,20 @@ class UserTokenAdmin(admin.ModelAdmin):
@admin.display(description=gettext_lazy("Is Expired"), boolean=True)
def is_expired(self, token: UserToken) -> bool:
return token.expired_at < timezone.now()


@admin.register(UserPasskeyCredential)
class UserPasskeyCredentialAdmin(admin.ModelAdmin):
list_display = [
"id",
"user",
"name",
"device_type",
"backup_state",
"is_enabled",
"last_used_at",
"created_at",
]
list_filter = ["is_enabled", "backup_state", "device_type"]
search_fields = ["user__username", "credential_id", "name"]
ordering = ["-id"]
4 changes: 4 additions & 0 deletions apps/account/constants.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,10 @@
WECHAT_LOGIN_STATE_KEY = "wechat_login_state:{state}"
WECHAT_USER_INFO_KEY = "wechat_user_info:{code}"

PASSKEY_AUTH_CHALLENGE_KEY = "passkey_auth_challenge:{challenge_id}"
PASSKEY_REGISTER_CHALLENGE_KEY = "passkey_register_challenge:{challenge_id}"
PASSKEY_BIND_KEY = "passkey_bind:{code}"


class UserTypeChoices(TextChoices):
"""
Expand Down
30 changes: 30 additions & 0 deletions apps/account/exceptions.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,11 @@ class WrongSignInParam(APIException):
default_detail = gettext_lazy("Wrong Username or Password")


class PasswordLoginDisabled(APIException):
status_code = status.HTTP_400_BAD_REQUEST
default_detail = gettext_lazy("Password Login Disabled")


class TokenNotExist(APIException):
status_code = status.HTTP_400_BAD_REQUEST
default_detail = gettext_lazy("Token Not Exist")
Expand Down Expand Up @@ -51,3 +56,28 @@ class PhoneVerifyCodeInvalid(APIException):
class RegistryLocked(APIException):
status_code = status.HTTP_503_SERVICE_UNAVAILABLE
default_detail = gettext_lazy("Registry Disabled")


class PasskeyAuthenticationFailed(APIException):
status_code = status.HTTP_400_BAD_REQUEST
default_detail = gettext_lazy("Passkey Authentication Failed")


class PasskeyRegistrationFailed(APIException):
status_code = status.HTTP_400_BAD_REQUEST
default_detail = gettext_lazy("Passkey Registration Failed")


class PasskeyCredentialNotFound(APIException):
status_code = status.HTTP_404_NOT_FOUND
default_detail = gettext_lazy("Passkey Credential Not Found")


class PasskeyCodeInvalid(APIException):
status_code = status.HTTP_400_BAD_REQUEST
default_detail = gettext_lazy("Passkey Code Invalid")


class PasskeyCredentialExists(APIException):
status_code = status.HTTP_400_BAD_REQUEST
default_detail = gettext_lazy("Passkey Credential Already Exists")
53 changes: 53 additions & 0 deletions apps/account/migrations/0009_userpasskeycredential.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,53 @@
# pylint: disable=C0103
# Generated by Django 5.2.13 on 2026-06-26 06:57

import django.db.models.deletion
from django.conf import settings
from django.db import migrations, models


class Migration(migrations.Migration):
dependencies = [
("account", "0008_rename_usertoken_user_session_key_expired_at_account_use_user_id_c02769_idx_and_more"),
]

operations = [
migrations.CreateModel(
name="UserPasskeyCredential",
fields=[
("id", models.BigAutoField(primary_key=True, serialize=False, verbose_name="ID")),
(
"credential_id",
models.CharField(db_index=True, max_length=512, unique=True, verbose_name="Credential ID"),
),
("public_key", models.BinaryField(verbose_name="Public Key")),
("sign_count", models.PositiveBigIntegerField(default=0, verbose_name="Sign Count")),
("user_handle", models.CharField(db_index=True, max_length=255, verbose_name="User Handle")),
("transports", models.JSONField(blank=True, default=list, verbose_name="Transports")),
("device_type", models.CharField(blank=True, max_length=64, null=True, verbose_name="Device Type")),
("backup_state", models.BooleanField(default=False, verbose_name="Backup State")),
("name", models.CharField(blank=True, max_length=128, null=True, verbose_name="Name")),
("last_used_at", models.DateTimeField(blank=True, null=True, verbose_name="Last Used at")),
("is_enabled", models.BooleanField(db_index=True, default=True, verbose_name="Is Enabled")),
("created_at", models.DateTimeField(auto_now_add=True, verbose_name="Created at")),
(
"user",
models.ForeignKey(
db_constraint=False,
on_delete=django.db.models.deletion.PROTECT,
to=settings.AUTH_USER_MODEL,
verbose_name="User",
),
),
],
options={
"verbose_name": "User Passkey Credential",
"verbose_name_plural": "User Passkey Credential",
"ordering": ["-id"],
"indexes": [
models.Index(fields=["user", "is_enabled"], name="account_use_user_id_bce0b4_idx"),
models.Index(fields=["user", "created_at"], name="account_use_user_id_40c1a8_idx"),
],
},
),
]
31 changes: 31 additions & 0 deletions apps/account/models.py
Original file line number Diff line number Diff line change
Expand Up @@ -217,3 +217,34 @@ class Meta:
models.Index(fields=["user", "expired_at"]),
models.Index(fields=["user", "session_key", "expired_at"]),
]


class UserPasskeyCredential(models.Model):
"""
User Passkey Credential
"""

id = models.BigAutoField(gettext_lazy("ID"), primary_key=True)
user = models.ForeignKey(
verbose_name=gettext_lazy("User"), to="User", on_delete=models.PROTECT, db_constraint=False
)
credential_id = models.CharField(gettext_lazy("Credential ID"), max_length=512, unique=True, db_index=True)
public_key = models.BinaryField(gettext_lazy("Public Key"))
sign_count = models.PositiveBigIntegerField(gettext_lazy("Sign Count"), default=0)
user_handle = models.CharField(gettext_lazy("User Handle"), max_length=255, db_index=True)
transports = models.JSONField(gettext_lazy("Transports"), default=list, blank=True)
device_type = models.CharField(gettext_lazy("Device Type"), max_length=64, blank=True, null=True)
backup_state = models.BooleanField(gettext_lazy("Backup State"), default=False)
name = models.CharField(gettext_lazy("Name"), max_length=128, blank=True, null=True)
last_used_at = models.DateTimeField(gettext_lazy("Last Used at"), null=True, blank=True)
is_enabled = models.BooleanField(gettext_lazy("Is Enabled"), default=True, db_index=True)
created_at = models.DateTimeField(gettext_lazy("Created at"), auto_now_add=True)

class Meta:
verbose_name = gettext_lazy("User Passkey Credential")
verbose_name_plural = verbose_name
ordering = ["-id"]
indexes = [
models.Index(fields=["user", "is_enabled"]),
models.Index(fields=["user", "created_at"]),
]
218 changes: 218 additions & 0 deletions apps/account/passkeys.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,218 @@
import base64
import json
import secrets
from dataclasses import dataclass

from django.conf import settings
from django.core.cache import cache
from django.db import IntegrityError
from django.utils import timezone
from webauthn import (
generate_authentication_options,
generate_registration_options,
verify_authentication_response,
verify_registration_response,
)
from webauthn.helpers import options_to_json
from webauthn.helpers.exceptions import WebAuthnException
from webauthn.helpers.structs import (
AuthenticatorSelectionCriteria,
ResidentKeyRequirement,
UserVerificationRequirement,
)

from apps.account.constants import (
PASSKEY_AUTH_CHALLENGE_KEY,
PASSKEY_BIND_KEY,
PASSKEY_REGISTER_CHALLENGE_KEY,
)
from apps.account.exceptions import (
PasskeyAuthenticationFailed,
PasskeyCodeInvalid,
PasskeyCredentialExists,
PasskeyCredentialNotFound,
PasskeyRegistrationFailed,
)
from apps.account.models import User, UserPasskeyCredential


@dataclass
class PasskeyOptions:
challenge_id: str
public_key: dict


def b64url_encode(value: bytes) -> str:
return base64.urlsafe_b64encode(value).decode().rstrip("=")


def b64url_decode(value: str) -> bytes:
return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))


def generate_passkey_login_options(session_key: str) -> PasskeyOptions:
options = generate_authentication_options(
rp_id=settings.PASSKEY_RP_ID,
timeout=settings.PASSKEY_CHALLENGE_TIMEOUT * 1000,
user_verification=UserVerificationRequirement.REQUIRED,
)
challenge_id = secrets.token_urlsafe(32)
cache.set(
PASSKEY_AUTH_CHALLENGE_KEY.format(challenge_id=challenge_id),
{"challenge": b64url_encode(options.challenge), "session_key": session_key},
timeout=settings.PASSKEY_CHALLENGE_TIMEOUT,
)
public_key = json.loads(options_to_json(options))
if not public_key.get("allowCredentials"):
public_key.pop("allowCredentials", None)
return PasskeyOptions(challenge_id=challenge_id, public_key=public_key)


def generate_passkey_register_options(session_key: str, display_name: str = "") -> PasskeyOptions:
user_handle = secrets.token_bytes(32)
user_name = display_name or "Pending Passkey User"
options = generate_registration_options(
rp_id=settings.PASSKEY_RP_ID,
rp_name=settings.PASSKEY_RP_NAME,
user_name=user_name,
user_id=user_handle,
user_display_name=user_name,
timeout=settings.PASSKEY_CHALLENGE_TIMEOUT * 1000,
authenticator_selection=AuthenticatorSelectionCriteria(
resident_key=ResidentKeyRequirement.REQUIRED,
require_resident_key=True,
user_verification=UserVerificationRequirement.REQUIRED,
),
)
challenge_id = secrets.token_urlsafe(32)
cache.set(
PASSKEY_REGISTER_CHALLENGE_KEY.format(challenge_id=challenge_id),
{
"challenge": b64url_encode(options.challenge),
"session_key": session_key,
"user_handle": b64url_encode(user_handle),
},
timeout=settings.PASSKEY_CHALLENGE_TIMEOUT,
)
return PasskeyOptions(challenge_id=challenge_id, public_key=json.loads(options_to_json(options)))


def create_pending_passkey(challenge_id: str, credential: dict, session_key: str, name: str = "") -> str:
cache_key = PASSKEY_REGISTER_CHALLENGE_KEY.format(challenge_id=challenge_id)
challenge_data = cache.get(cache_key)
cache.delete(cache_key)
if not challenge_data or challenge_data.get("session_key") != session_key:
raise PasskeyRegistrationFailed()

try:
verified = verify_registration_response(
credential=credential,
expected_challenge=b64url_decode(challenge_data["challenge"]),
expected_rp_id=settings.PASSKEY_RP_ID,
expected_origin=settings.PASSKEY_ALLOWED_ORIGINS,
require_user_verification=True,
)
except (KeyError, TypeError, ValueError, WebAuthnException) as err:
raise PasskeyRegistrationFailed() from err

credential_id = b64url_encode(verified.credential_id)
# pylint: disable=E1101
if UserPasskeyCredential.objects.filter(credential_id=credential_id).exists():
raise PasskeyCredentialExists()

code = secrets.token_urlsafe(32)
cache.set(
PASSKEY_BIND_KEY.format(code=code),
{
"credential_id": credential_id,
"public_key": b64url_encode(verified.credential_public_key),
"session_key": session_key,
"sign_count": verified.sign_count,
"user_handle": challenge_data["user_handle"],
"transports": _load_transports(credential),
"device_type": getattr(verified.credential_device_type, "value", str(verified.credential_device_type)),
"backup_state": verified.credential_backed_up,
"name": name or "",
},
timeout=settings.PASSKEY_BIND_TIMEOUT,
)
return code


def bind_pending_passkey(user: User, code: str, session_key: str) -> UserPasskeyCredential:
cache_key = PASSKEY_BIND_KEY.format(code=code)
pending = cache.get(cache_key)
cache.delete(cache_key)
if not pending or pending.get("session_key") != session_key:
raise PasskeyCodeInvalid()

try:
# pylint: disable=E1101
return UserPasskeyCredential.objects.create(
user=user,
credential_id=pending["credential_id"],
public_key=b64url_decode(pending["public_key"]),
sign_count=pending["sign_count"],
user_handle=pending["user_handle"],
transports=pending["transports"],
device_type=pending["device_type"],
backup_state=pending["backup_state"],
name=pending["name"] or None,
)
except IntegrityError as err:
raise PasskeyCredentialExists() from err


def verify_passkey_login(challenge_id: str, credential: dict, session_key: str) -> User:
cache_key = PASSKEY_AUTH_CHALLENGE_KEY.format(challenge_id=challenge_id)
challenge_data = cache.get(cache_key)
cache.delete(cache_key)
if not challenge_data or challenge_data.get("session_key") != session_key:
raise PasskeyAuthenticationFailed()

credential_id = _load_credential_id(credential)
# pylint: disable=E1101
passkey = (
UserPasskeyCredential.objects.filter(credential_id=credential_id, is_enabled=True)
.select_related("user")
.first()
)
if not passkey:
raise PasskeyCredentialNotFound()

try:
verified = verify_authentication_response(
credential=credential,
expected_challenge=b64url_decode(challenge_data["challenge"]),
expected_rp_id=settings.PASSKEY_RP_ID,
expected_origin=settings.PASSKEY_ALLOWED_ORIGINS,
credential_public_key=bytes(passkey.public_key),
credential_current_sign_count=passkey.sign_count,
require_user_verification=True,
)
except (KeyError, TypeError, ValueError, WebAuthnException) as err:
raise PasskeyAuthenticationFailed() from err

if b64url_encode(verified.credential_id) != passkey.credential_id or not verified.user_verified:
raise PasskeyAuthenticationFailed()

passkey.sign_count = verified.new_sign_count
passkey.device_type = getattr(verified.credential_device_type, "value", str(verified.credential_device_type))
passkey.backup_state = verified.credential_backed_up
passkey.last_used_at = timezone.now()
passkey.save(update_fields=["sign_count", "device_type", "backup_state", "last_used_at"])
return passkey.user


def _load_credential_id(credential: dict) -> str:
credential_id = credential.get("id") or credential.get("rawId")
if not credential_id:
raise PasskeyAuthenticationFailed()
return credential_id


def _load_transports(credential: dict) -> list[str]:
transports = credential.get("response", {}).get("transports") or []
if not isinstance(transports, list):
return []
return [str(transport) for transport in transports]
Loading
Loading