mirror of
https://gitlab.com/crafty-controller/crafty-4.git
synced 2026-08-19 20:23:07 -04:00
296 lines
11 KiB
Python
296 lines
11 KiB
Python
import base64
|
|
import json
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
|
|
from webauthn import (
|
|
generate_registration_options,
|
|
verify_registration_response,
|
|
generate_authentication_options,
|
|
verify_authentication_response,
|
|
options_to_json,
|
|
)
|
|
from webauthn.helpers.structs import (
|
|
AuthenticatorSelectionCriteria,
|
|
ResidentKeyRequirement,
|
|
UserVerificationRequirement,
|
|
PublicKeyCredentialDescriptor,
|
|
)
|
|
from webauthn.helpers.cose import COSEAlgorithmIdentifier
|
|
|
|
from app.classes.helpers.helpers import Helpers
|
|
from app.classes.models.users import HelperUsers
|
|
from app.classes.models.passkey import HelperPasskey, PasskeyData
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
CHALLENGE_TIMEOUT_MINUTES = 5
|
|
|
|
|
|
def _utc_now():
|
|
"""Return current UTC time as a naive datetime (for database compatibility)."""
|
|
return datetime.now(timezone.utc).replace(tzinfo=None)
|
|
|
|
|
|
class PasskeyController:
|
|
def __init__(self, passkey_helper, helper):
|
|
self.passkey_helper: HelperPasskey = passkey_helper
|
|
self.helper = helper
|
|
|
|
def get_rp_id(self):
|
|
base_url = self.helper.get_setting("base_url", "localhost:8443")
|
|
return base_url.split(":")[0]
|
|
|
|
def get_rp_name(self):
|
|
return self.helper.get_setting("passkey_rp_name", "Crafty Controller")
|
|
|
|
def get_origin(self):
|
|
base_url = self.helper.get_setting("base_url", "localhost:8443")
|
|
return f"https://{base_url}"
|
|
|
|
def is_enabled(self):
|
|
return self.helper.get_setting("enable_passkey_auth", False)
|
|
|
|
def generate_registration_options(self, user_id):
|
|
user = HelperUsers.get_user(user_id)
|
|
|
|
existing_passkeys = list(self.passkey_helper.get_user_passkeys(user_id))
|
|
exclude_credentials = [
|
|
PublicKeyCredentialDescriptor(id=pk.credential_id)
|
|
for pk in existing_passkeys
|
|
]
|
|
|
|
options = generate_registration_options(
|
|
rp_id=self.get_rp_id(),
|
|
rp_name=self.get_rp_name(),
|
|
user_id=str(user_id).encode(),
|
|
user_name=user["username"],
|
|
user_display_name=user["username"],
|
|
exclude_credentials=exclude_credentials if exclude_credentials else None,
|
|
authenticator_selection=AuthenticatorSelectionCriteria(
|
|
resident_key=ResidentKeyRequirement.PREFERRED,
|
|
user_verification=UserVerificationRequirement.PREFERRED,
|
|
),
|
|
supported_pub_key_algs=[
|
|
COSEAlgorithmIdentifier.ECDSA_SHA_256,
|
|
COSEAlgorithmIdentifier.RSASSA_PKCS1_v1_5_SHA_256,
|
|
],
|
|
timeout=CHALLENGE_TIMEOUT_MINUTES * 60 * 1000,
|
|
)
|
|
|
|
challenge_id = Helpers.create_uuid()
|
|
expires_at = _utc_now() + timedelta(minutes=CHALLENGE_TIMEOUT_MINUTES)
|
|
|
|
user_obj = HelperUsers.get_by_id(user_id)
|
|
self.passkey_helper.store_challenge(
|
|
challenge_id=challenge_id,
|
|
user=user_obj,
|
|
challenge_bytes=options.challenge,
|
|
challenge_type="registration",
|
|
expires_at=expires_at,
|
|
)
|
|
|
|
logger.info("Generated passkey registration options for user %s", user_id)
|
|
|
|
return {
|
|
"options": json.loads(options_to_json(options)),
|
|
"challenge_id": challenge_id,
|
|
}
|
|
|
|
def verify_registration(self, user_id, challenge_id, credential_name, response):
|
|
challenge_record = self.passkey_helper.get_challenge(challenge_id)
|
|
if not challenge_record:
|
|
logger.warning("Invalid or expired challenge ID: %s", challenge_id)
|
|
return False
|
|
|
|
if challenge_record.challenge_type != "registration":
|
|
logger.warning("Wrong challenge type for registration: %s", challenge_id)
|
|
self.passkey_helper.delete_challenge(challenge_id)
|
|
return False
|
|
|
|
if challenge_record.user.user_id != user_id:
|
|
logger.warning("Challenge user mismatch for %s", challenge_id)
|
|
return False
|
|
|
|
if challenge_record.expires_at <= _utc_now():
|
|
logger.warning("Expired challenge: %s", challenge_id)
|
|
self.passkey_helper.delete_challenge(challenge_id)
|
|
return False
|
|
|
|
try:
|
|
verification = verify_registration_response(
|
|
credential=response,
|
|
expected_challenge=challenge_record.challenge,
|
|
expected_rp_id=self.get_rp_id(),
|
|
expected_origin=self.get_origin(),
|
|
)
|
|
except Exception as e:
|
|
logger.exception("Registration verification failed: %s", e)
|
|
self.passkey_helper.delete_challenge(challenge_id)
|
|
return False
|
|
|
|
self.passkey_helper.delete_challenge(challenge_id)
|
|
|
|
user_obj = HelperUsers.get_by_id(user_id)
|
|
passkey_id = Helpers.create_uuid()
|
|
|
|
# Get transports from input credential, not verification output
|
|
transports = response.get("response", {}).get("transports", [])
|
|
transports_json = json.dumps(transports)
|
|
|
|
passkey = self.passkey_helper.create_passkey(
|
|
passkey_id=passkey_id,
|
|
name=credential_name or "Passkey",
|
|
user=user_obj,
|
|
credential_id=verification.credential_id,
|
|
public_key=verification.credential_public_key,
|
|
sign_count=verification.sign_count,
|
|
transports=transports_json,
|
|
device_type=verification.credential_device_type.value,
|
|
backed_up=verification.credential_backed_up,
|
|
)
|
|
|
|
logger.info(
|
|
"Successfully registered passkey %s for user %s", passkey_id, user_id
|
|
)
|
|
return passkey
|
|
|
|
def generate_authentication_options(self, username=None):
|
|
allow_credentials = []
|
|
user_obj = None
|
|
|
|
if username:
|
|
user_id = HelperUsers.get_user_id_by_name(username)
|
|
if user_id:
|
|
user_obj = HelperUsers.get_by_id(user_id)
|
|
existing_passkeys = list(self.passkey_helper.get_user_passkeys(user_id))
|
|
|
|
# Don't return early if no passkeys -
|
|
# proceed with empty allow_credentials
|
|
# to prevent user enumeration. Auth will fail at verify step.
|
|
allow_credentials = [
|
|
PublicKeyCredentialDescriptor(
|
|
id=pk.credential_id,
|
|
transports=(
|
|
json.loads(pk.transports)
|
|
if pk.transports and pk.transports not in ("", "[]")
|
|
else None
|
|
),
|
|
)
|
|
for pk in existing_passkeys
|
|
]
|
|
|
|
options = generate_authentication_options(
|
|
rp_id=self.get_rp_id(),
|
|
allow_credentials=allow_credentials if allow_credentials else None,
|
|
user_verification=UserVerificationRequirement.PREFERRED,
|
|
timeout=CHALLENGE_TIMEOUT_MINUTES * 60 * 1000,
|
|
)
|
|
|
|
challenge_id = Helpers.create_uuid()
|
|
expires_at = _utc_now() + timedelta(minutes=CHALLENGE_TIMEOUT_MINUTES)
|
|
|
|
self.passkey_helper.store_challenge(
|
|
challenge_id=challenge_id,
|
|
user=user_obj,
|
|
challenge_bytes=options.challenge,
|
|
challenge_type="authentication",
|
|
expires_at=expires_at,
|
|
)
|
|
|
|
logger.info("Generated passkey authentication options")
|
|
|
|
return {
|
|
"options": json.loads(options_to_json(options)),
|
|
"challenge_id": challenge_id,
|
|
}
|
|
|
|
def verify_authentication(self, challenge_id, response):
|
|
challenge_record = self.passkey_helper.get_challenge(challenge_id)
|
|
if not challenge_record:
|
|
logger.warning("Invalid challenge ID: %s", challenge_id)
|
|
return False
|
|
|
|
if challenge_record.challenge_type != "authentication":
|
|
logger.warning("Wrong challenge type for authentication: %s", challenge_id)
|
|
self.passkey_helper.delete_challenge(challenge_id)
|
|
return False
|
|
|
|
if challenge_record.expires_at <= _utc_now():
|
|
logger.warning("Expired challenge: %s", challenge_id)
|
|
self.passkey_helper.delete_challenge(challenge_id)
|
|
return False
|
|
|
|
credential_id = response.get("rawId")
|
|
|
|
try:
|
|
# Proper base64url padding calculation
|
|
padding = 4 - (len(credential_id) % 4)
|
|
if padding != 4:
|
|
credential_id += "=" * padding
|
|
credential_id_bytes = base64.urlsafe_b64decode(credential_id)
|
|
except Exception:
|
|
logger.warning("Invalid credential ID encoding")
|
|
self.passkey_helper.delete_challenge(challenge_id)
|
|
return False
|
|
|
|
passkey = self.passkey_helper.get_passkey_by_credential_id(credential_id_bytes)
|
|
if not passkey:
|
|
logger.warning("Unknown credential ID")
|
|
self.passkey_helper.delete_challenge(challenge_id)
|
|
return False
|
|
|
|
if (
|
|
challenge_record.user
|
|
and challenge_record.user.user_id != passkey.user.user_id
|
|
):
|
|
logger.warning("Credential user mismatch")
|
|
self.passkey_helper.delete_challenge(challenge_id)
|
|
return False
|
|
|
|
try:
|
|
verification = verify_authentication_response(
|
|
credential=response,
|
|
expected_challenge=challenge_record.challenge,
|
|
expected_rp_id=self.get_rp_id(),
|
|
expected_origin=self.get_origin(),
|
|
credential_public_key=passkey.public_key,
|
|
credential_current_sign_count=passkey.sign_count,
|
|
)
|
|
except Exception as e:
|
|
logger.exception("Authentication verification failed: %s", e)
|
|
self.passkey_helper.delete_challenge(challenge_id)
|
|
return False
|
|
|
|
self.passkey_helper.delete_challenge(challenge_id)
|
|
|
|
# Always update last_used_at on successful authentication
|
|
self.passkey_helper.update_last_used(passkey.id)
|
|
|
|
if verification.new_sign_count > passkey.sign_count:
|
|
self.passkey_helper.update_sign_count(
|
|
passkey.id, verification.new_sign_count
|
|
)
|
|
elif (
|
|
verification.new_sign_count > 0
|
|
and verification.new_sign_count <= passkey.sign_count
|
|
):
|
|
logger.warning(
|
|
"Possible cloned authenticator detected for passkey %s", passkey.id
|
|
)
|
|
|
|
logger.info(
|
|
"Successfully authenticated user %s via passkey", passkey.user.user_id
|
|
)
|
|
return passkey.user.user_id
|
|
|
|
def delete_passkey(self, passkey_id, user_id):
|
|
passkey = PasskeyData.get_or_none(PasskeyData.id == passkey_id)
|
|
if not passkey or passkey.user.user_id != user_id:
|
|
return False
|
|
return self.passkey_helper.delete_passkey(passkey_id)
|
|
|
|
def purge_expired_challenges(self):
|
|
logger.info("Purging expired passkey challenges")
|
|
self.passkey_helper.cleanup_expired_challenges()
|