✨ Advanced Authentication
This commit is contained in:
Mawoka
@@ -0,0 +1,206 @@
|
||||
# This Source Code Form is subject to the terms of the Mozilla Public
|
||||
# License, v. 2.0. If a copy of the MPL was not distributed with this
|
||||
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
|
||||
import base64
|
||||
import enum
|
||||
import os
|
||||
import uuid
|
||||
|
||||
import pyotp
|
||||
from fastapi import APIRouter, HTTPException, Request, Response
|
||||
from pydantic import BaseModel, ValidationError
|
||||
|
||||
from classquiz.auth import verify_password
|
||||
from classquiz.config import redis, settings
|
||||
|
||||
from classquiz.db.models import User, FidoCredentials
|
||||
from webauthn import (
|
||||
generate_authentication_options,
|
||||
options_to_json,
|
||||
verify_authentication_response,
|
||||
base64url_to_bytes,
|
||||
)
|
||||
from webauthn.helpers.structs import (
|
||||
PublicKeyCredentialDescriptor,
|
||||
UserVerificationRequirement,
|
||||
AuthenticationCredential,
|
||||
)
|
||||
|
||||
from classquiz.oauth.authenticate_user import log_user_in
|
||||
|
||||
settings = settings()
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
class StartLoginInput(BaseModel):
|
||||
email: str
|
||||
|
||||
|
||||
class StartLoginResponseTypes(enum.Enum):
|
||||
PASSWORD = "PASSWORD"
|
||||
PASSKEY = "PASSKEY"
|
||||
TOTP = "TOTP"
|
||||
BACKUP = "BACKUP"
|
||||
|
||||
|
||||
class LoginSession(BaseModel):
|
||||
user_id: str
|
||||
step_1: set[StartLoginResponseTypes]
|
||||
step_2: set[StartLoginResponseTypes]
|
||||
webauthn_challenge: str | None
|
||||
step1_success: bool = False
|
||||
|
||||
|
||||
class StartLoginResponse(BaseModel):
|
||||
step_1: set[StartLoginResponseTypes]
|
||||
step_2: set[StartLoginResponseTypes]
|
||||
session_id: str
|
||||
webauthn_data: None | str
|
||||
|
||||
|
||||
def verify_webauthn(data, fidocredentialss: list[FidoCredentials], login_session: LoginSession):
|
||||
try:
|
||||
credential = AuthenticationCredential.parse_obj(data)
|
||||
except ValidationError:
|
||||
print("ValidationError")
|
||||
raise HTTPException(401)
|
||||
|
||||
user_cred: FidoCredentials | None = None
|
||||
|
||||
credential.id = base64url_to_bytes(credential.id)
|
||||
for cred in fidocredentialss:
|
||||
if cred.id == credential.id:
|
||||
user_cred = cred
|
||||
break
|
||||
if user_cred is None:
|
||||
print("user_cred not in DB")
|
||||
raise HTTPException(401)
|
||||
credential.id = base64.urlsafe_b64encode(credential.raw_id).decode("utf-8").replace("=", "")
|
||||
credential.response.client_data_json = base64.b64decode(credential.response.client_data_json + b"==")
|
||||
credential.response.authenticator_data = base64.urlsafe_b64decode(credential.response.authenticator_data + b"==")
|
||||
credential.response.signature = base64.urlsafe_b64decode(credential.response.signature + b"==")
|
||||
print(credential)
|
||||
try:
|
||||
verify_authentication_response(
|
||||
credential=credential,
|
||||
expected_challenge=base64.b64decode(login_session.webauthn_challenge),
|
||||
expected_rp_id="localhost",
|
||||
expected_origin=settings.root_address,
|
||||
credential_public_key=user_cred.public_key,
|
||||
credential_current_sign_count=user_cred.sign_count,
|
||||
require_user_verification=False,
|
||||
)
|
||||
print("logging in...")
|
||||
return True
|
||||
except Exception as err:
|
||||
print(err)
|
||||
raise HTTPException(status_code=401)
|
||||
|
||||
|
||||
@router.post("/start")
|
||||
async def start_login(data: StartLoginInput):
|
||||
user = await User.objects.select_related("fidocredentialss").get_or_none(email=data.email)
|
||||
step_1: set[StartLoginResponseTypes] = set()
|
||||
step_2: set[StartLoginResponseTypes] = set()
|
||||
webauthn_data = None
|
||||
webauthn_challenge = None
|
||||
if user is None:
|
||||
step_1.add(StartLoginResponseTypes.PASSWORD)
|
||||
return StartLoginResponse(step_1=step_1, step_2=step_2, session_id=os.urandom(16).hex(), webauthn_data=None)
|
||||
if user.password is not None:
|
||||
step_1.add(StartLoginResponseTypes.PASSWORD)
|
||||
if len(user.fidocredentialss) > 0:
|
||||
if user.require_password is True:
|
||||
step_2.add(StartLoginResponseTypes.PASSKEY)
|
||||
else:
|
||||
step_1.add(StartLoginResponseTypes.PASSKEY)
|
||||
webauthn_data = generate_authentication_options(
|
||||
rp_id="localhost",
|
||||
allow_credentials=[
|
||||
PublicKeyCredentialDescriptor(id=cred.id, type="public-key") for cred in user.fidocredentialss
|
||||
],
|
||||
user_verification=UserVerificationRequirement.PREFERRED,
|
||||
)
|
||||
webauthn_challenge = base64.b64encode(webauthn_data.challenge).decode("utf-8")
|
||||
webauthn_data = options_to_json(webauthn_data)
|
||||
if user.totp_secret is not None:
|
||||
if user.require_password:
|
||||
step_2.add(StartLoginResponseTypes.TOTP)
|
||||
else:
|
||||
step_1.add(StartLoginResponseTypes.TOTP)
|
||||
login_session = LoginSession(
|
||||
step_1=step_1,
|
||||
step_2=step_2,
|
||||
user_id=user.id.hex,
|
||||
webauthn_challenge=webauthn_challenge,
|
||||
)
|
||||
session_id = os.urandom(16).hex()
|
||||
await redis.set(f"login_session:{session_id}", login_session.json(), ex=600)
|
||||
return StartLoginResponse(step_1=step_1, step_2=step_2, session_id=session_id, webauthn_data=webauthn_data)
|
||||
|
||||
|
||||
class StepInput(BaseModel):
|
||||
auth_type: StartLoginResponseTypes
|
||||
data: str | dict
|
||||
|
||||
|
||||
@router.post("/step/{step_id}")
|
||||
async def step_1(session_id: str, data: StepInput, request: Request, response: Response, step_id: int):
|
||||
if step_id < 0 or step_id > 2:
|
||||
raise HTTPException(status_code=401)
|
||||
redis_res = await redis.get(f"login_session:{session_id}")
|
||||
if redis_res is None:
|
||||
raise HTTPException(401, detail="wrong credentials")
|
||||
login_session = LoginSession.parse_raw(redis_res)
|
||||
|
||||
if step_id == 1:
|
||||
if data.auth_type not in {*login_session.step_1, StartLoginResponseTypes.BACKUP}:
|
||||
print("AUTH_TYPE not in data")
|
||||
raise HTTPException(401)
|
||||
elif step_id == 2:
|
||||
if data.auth_type not in login_session.step_2:
|
||||
print("AUTH_TYPE not in data")
|
||||
raise HTTPException(401)
|
||||
else:
|
||||
print("unknown step")
|
||||
raise HTTPException(401)
|
||||
user = await User.objects.select_related("fidocredentialss").get_or_none(id=uuid.UUID(login_session.user_id))
|
||||
if data.auth_type == StartLoginResponseTypes.PASSWORD:
|
||||
if verify_password(data.data, user.password):
|
||||
if len(login_session.step_2) == 0 or (step_id == 2 and login_session.step1_success is True):
|
||||
return await log_user_in(user, request, response)
|
||||
else:
|
||||
login_session.step1_success = True
|
||||
await redis.set(f"login_session:{session_id}", login_session.json(), ex=600)
|
||||
return Response(status_code=202)
|
||||
else:
|
||||
print("Wrong Password")
|
||||
raise HTTPException(401, detail="wrong credentials")
|
||||
elif data.auth_type == StartLoginResponseTypes.PASSKEY:
|
||||
res = verify_webauthn(data=data.data, fidocredentialss=user.fidocredentialss, login_session=login_session)
|
||||
if res is True:
|
||||
if len(login_session.step_2) == 0 or (step_id == 2 and login_session.step1_success is True):
|
||||
return await log_user_in(user, request, response)
|
||||
else:
|
||||
login_session.step1_success = True
|
||||
await redis.set(f"login_session:{session_id}", login_session.json(), ex=600)
|
||||
return Response(status_code=202)
|
||||
else:
|
||||
raise HTTPException(401, detail="webauthn failed")
|
||||
elif data.auth_type == StartLoginResponseTypes.BACKUP:
|
||||
if user.backup_code == data.data:
|
||||
user.backup_code = os.urandom(32).hex()
|
||||
await user.update()
|
||||
return await log_user_in(user, request, response)
|
||||
else:
|
||||
print("Wrong Backup-Code")
|
||||
raise HTTPException(status_code=401)
|
||||
elif data.auth_type == StartLoginResponseTypes.TOTP:
|
||||
if step_id == 1 and user.require_password:
|
||||
print("TOTP Cant be step 1")
|
||||
raise HTTPException(401)
|
||||
totp = pyotp.TOTP(user.totp_secret)
|
||||
if totp.verify(data.data):
|
||||
return await log_user_in(user, request, response)
|
||||
else:
|
||||
raise HTTPException(401, detail="totp wrong")
|
||||
@@ -1,6 +1,7 @@
|
||||
# This Source Code Form is subject to the terms of the Mozilla Public
|
||||
# License, v. 2.0. If a copy of the MPL was not distributed with this
|
||||
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
|
||||
|
||||
import gzip
|
||||
import os
|
||||
|
||||
@@ -33,10 +34,14 @@ import bleach
|
||||
from pydantic import BaseModel
|
||||
from classquiz.db.models import User, UserSession, UpdatePassword, Token, Quiz, ApiKey
|
||||
from classquiz.emails import send_register_email, send_forgotten_password_email
|
||||
from classquiz.routers.users import webauthn, twofa
|
||||
|
||||
settings = settings()
|
||||
router = APIRouter()
|
||||
|
||||
router.include_router(webauthn.router, prefix="/webauthn")
|
||||
router.include_router(twofa.router, prefix="/2fa")
|
||||
|
||||
|
||||
class RouteUser(pydantic.BaseModel):
|
||||
username: str
|
||||
@@ -84,14 +89,15 @@ async def create_user(user: RouteUser, background_task: BackgroundTasks) -> User
|
||||
return user
|
||||
|
||||
|
||||
@router.post("/token/cookie", response_model=Token)
|
||||
@router.post("/token/cookie", response_model=Token, deprecated=True)
|
||||
async def login_for_cookie_access_token(
|
||||
request: Request,
|
||||
response: Response,
|
||||
form_data: OAuth2PasswordRequestForm = Depends(),
|
||||
):
|
||||
user = await authenticate_user(form_data.username, form_data.password)
|
||||
if not user:
|
||||
user = await User.objects.select_related("fidocredentialss").get(id=user.id)
|
||||
if not user or user.totp_secret is not None or user or len(user.fidocredentialss) != 0:
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||||
detail="Incorrect username or password",
|
||||
@@ -169,7 +175,17 @@ async def signout_everywhere(response: Response, user: User = Depends(get_curren
|
||||
|
||||
@router.get(
|
||||
"/me",
|
||||
response_model_exclude={"password", "verify_key", "usersessions", "avatar", "google_uid", "quizs"},
|
||||
response_model_exclude={
|
||||
"password",
|
||||
"verify_key",
|
||||
"usersessions",
|
||||
"avatar",
|
||||
"quizs",
|
||||
"fidocredentialss",
|
||||
"backup_code",
|
||||
"apikeys",
|
||||
"totp_secret",
|
||||
},
|
||||
response_model=User,
|
||||
)
|
||||
async def get_me(user: User = Depends(get_current_user)):
|
||||
@@ -0,0 +1,75 @@
|
||||
# This Source Code Form is subject to the terms of the Mozilla Public
|
||||
# License, v. 2.0. If a copy of the MPL was not distributed with this
|
||||
# file, You can obtain one at https://mozilla.org/MPL/2.0/.
|
||||
import os
|
||||
import urllib.parse
|
||||
|
||||
import pyotp
|
||||
from fastapi import APIRouter, Depends
|
||||
from pydantic import BaseModel
|
||||
|
||||
from classquiz.auth import get_current_user
|
||||
from classquiz.db.models import User
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
class GetBackupCodeResponse(BaseModel):
|
||||
code: str
|
||||
|
||||
|
||||
@router.get("/backup_code", response_model=GetBackupCodeResponse)
|
||||
async def get_backup_code(user: User = Depends(get_current_user)):
|
||||
backup_code = os.urandom(32).hex()
|
||||
user = await User.objects.get(id=user.id)
|
||||
user.backup_code = backup_code
|
||||
await user.update()
|
||||
return GetBackupCodeResponse(code=backup_code)
|
||||
|
||||
|
||||
class SetRequirePassword(BaseModel):
|
||||
require_password: bool
|
||||
|
||||
|
||||
@router.post("/require_password", response_model=SetRequirePassword)
|
||||
async def set_require_password(data: SetRequirePassword, user: User = Depends(get_current_user)):
|
||||
user = await User.objects.get(id=user.id)
|
||||
user.require_password = data.require_password
|
||||
await user.update()
|
||||
return data
|
||||
|
||||
|
||||
class SetTotpUpResponse(BaseModel):
|
||||
url: str
|
||||
secret: str
|
||||
|
||||
|
||||
@router.post("/totp", response_model=SetTotpUpResponse)
|
||||
async def set_totp_up(user: User = Depends(get_current_user)):
|
||||
user = await User.objects.get(id=user.id)
|
||||
user.totp_secret = pyotp.random_base32()
|
||||
url = pyotp.totp.TOTP(user.totp_secret).provisioning_uri(
|
||||
name=urllib.parse.quote(user.username), issuer_name="ClassQuiz"
|
||||
)
|
||||
await user.update()
|
||||
return SetTotpUpResponse(url=url, secret=user.totp_secret)
|
||||
|
||||
|
||||
class GetTotpStatusResponse(BaseModel):
|
||||
activated: bool
|
||||
|
||||
|
||||
@router.get("/totp", response_model=GetTotpStatusResponse)
|
||||
async def get_totp_status(user: User = Depends(get_current_user)):
|
||||
user = await User.objects.get(id=user.id)
|
||||
if user.totp_secret is None:
|
||||
return GetTotpStatusResponse(activated=False)
|
||||
else:
|
||||
return GetTotpStatusResponse(activated=True)
|
||||
|
||||
|
||||
@router.delete("/totp")
|
||||
async def disable_totp(user: User = Depends(get_current_user)):
|
||||
user = await User.objects.get(id=user.id)
|
||||
user.totp_secret = None
|
||||
await user.update()
|
||||
@@ -0,0 +1,87 @@
|
||||
from fastapi import APIRouter, Depends, HTTPException
|
||||
from pydantic import BaseModel
|
||||
from webauthn.helpers.cose import COSEAlgorithmIdentifier
|
||||
|
||||
from classquiz.auth import get_current_user
|
||||
from classquiz.db.models import User, FidoCredentials
|
||||
from classquiz.config import redis, settings
|
||||
|
||||
import base64
|
||||
|
||||
from webauthn import generate_registration_options, verify_registration_response
|
||||
from webauthn.helpers.structs import (
|
||||
AuthenticatorSelectionCriteria,
|
||||
UserVerificationRequirement,
|
||||
RegistrationCredential,
|
||||
PublicKeyCredentialCreationOptions,
|
||||
PublicKeyCredentialDescriptor,
|
||||
)
|
||||
|
||||
settings = settings()
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
@router.get("/add_key", response_model=PublicKeyCredentialCreationOptions)
|
||||
async def request_add_key_data(user: User = Depends(get_current_user)):
|
||||
user = await User.objects.select_related("fidocredentialss").get(id=user.id)
|
||||
options = generate_registration_options(
|
||||
rp_id="localhost",
|
||||
rp_name="ClassQuiz",
|
||||
user_id=user.id.hex,
|
||||
user_name=user.email,
|
||||
exclude_credentials=[
|
||||
PublicKeyCredentialDescriptor(id=cred.id, type="public-key", transports=[])
|
||||
for cred in user.fidocredentialss
|
||||
],
|
||||
authenticator_selection=AuthenticatorSelectionCriteria(user_verification=UserVerificationRequirement.PREFERRED),
|
||||
supported_pub_key_algs=[
|
||||
COSEAlgorithmIdentifier.ECDSA_SHA_256,
|
||||
COSEAlgorithmIdentifier.RSASSA_PKCS1_v1_5_SHA_256,
|
||||
],
|
||||
timeout=600,
|
||||
)
|
||||
await redis.set(f"add_webauthn:{user.id.hex}", base64.b64encode(options.challenge), ex=610)
|
||||
return options
|
||||
|
||||
|
||||
@router.post("/add_key")
|
||||
async def confirm_add_key_data(credential: RegistrationCredential, user: User = Depends(get_current_user)):
|
||||
redis_res = await redis.get(f"add_webauthn:{user.id.hex}")
|
||||
if redis_res is None:
|
||||
raise HTTPException(401)
|
||||
current_registration_challenge = base64.b64decode(redis_res)
|
||||
credential.id = base64.urlsafe_b64encode(credential.raw_id).decode("utf-8").replace("=", "")
|
||||
credential.response.client_data_json = base64.b64decode(credential.response.client_data_json + b"==")
|
||||
credential.response.attestation_object = base64.urlsafe_b64decode(credential.response.attestation_object + b"==")
|
||||
verification = verify_registration_response(
|
||||
credential=credential,
|
||||
expected_challenge=current_registration_challenge,
|
||||
expected_rp_id="localhost",
|
||||
expected_origin=settings.root_address,
|
||||
)
|
||||
new_credential = FidoCredentials(
|
||||
id=verification.credential_id,
|
||||
public_key=verification.credential_public_key,
|
||||
sign_count=verification.sign_count,
|
||||
)
|
||||
user = await User.objects.select_related("fidocredentialss").get(id=user.id)
|
||||
await user.fidocredentialss.add(new_credential)
|
||||
|
||||
|
||||
class SecurityKey(BaseModel):
|
||||
id: int
|
||||
|
||||
|
||||
@router.get("/list", response_model=list[SecurityKey])
|
||||
async def list_security_keys(user: User = Depends(get_current_user)):
|
||||
user = await User.objects.select_related("fidocredentialss").get(id=user.id)
|
||||
return [SecurityKey(id=sec.pk) for sec in user.fidocredentialss]
|
||||
|
||||
|
||||
@router.delete("/key/{key_id}")
|
||||
async def delete_security_key(key_id: int, user: User = Depends(get_current_user)):
|
||||
key = await FidoCredentials.objects.get_or_none(pk=key_id, user=user.id)
|
||||
if key is None:
|
||||
raise HTTPException(status_code=404, detail="Key not found")
|
||||
await key.delete()
|
||||
Reference in New Issue
Block a user