🔒 Fixed some security issues. Please upgrade!

classquiz/routers/editor.py

@@ -72,10 +72,10 @@ async def finish_edit(edit_id: str, quiz_input: QuizInput):
     if session_data is None:
         raise HTTPException(status_code=401, detail="Edit ID not found!")
     session_data = EditSessionData.parse_raw(session_data)
-    quiz_input.title = html.unescape(bleach.clean(quiz_input.title, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True))
-    quiz_input.description = html.unescape(bleach.clean(quiz_input.description, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True))
+    quiz_input.title = bleach.clean(quiz_input.title, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True)
+    quiz_input.description = bleach.clean(quiz_input.description, tags=ALLOWED_TAGS_FOR_QUIZ, strip=True)
     if quiz_input.background_color is not None:
-        quiz_input.background_color = html.unescape(bleach.clean(quiz_input.background_color, tags=[], strip=True))
+        quiz_input.background_color = bleach.clean(quiz_input.background_color, tags=[], strip=True)
 
     for i, question in enumerate(quiz_input.questions):
         if question.type == QuizQuestionType.ABCD:

classquiz/routers/quiztivity/__init__.py

@@ -17,13 +17,13 @@ router = APIRouter()
 router.include_router(shares_router, prefix="/shares")
 
 
-@router.post("/create")
+@router.post("/create", response_model_exclude={"user": ...})
 async def create_quiztivity(data: QuizTivityInput, user: User = Depends(get_current_user)) -> QuizTivity:
     quiztivity = QuizTivity.parse_obj({**data.dict(), "user": user, "id": uuid4(), "created_at": datetime.now()})
     return await quiztivity.save()
 
 
-@router.get("/{uuid}")
+@router.get("/{uuid}", response_model_exclude={"user": ...})
 async def get_quiztivity(uuid: UUID) -> QuizTivity:
     quiztivity = await QuizTivity.objects.get_or_none(id=uuid)
     if quiztivity is None:
@@ -31,7 +31,7 @@ async def get_quiztivity(uuid: UUID) -> QuizTivity:
     return quiztivity
 
 
-@router.put("/{uuid}")
+@router.put("/{uuid}", response_model_exclude={"user": ...})
 async def put_quiztivity(data: QuizTivityInput, uuid: UUID, user: User = Depends(get_current_user)) -> QuizTivity:
     quiztivity = await QuizTivity.objects.get_or_none(id=uuid, user=user)
     if quiztivity is None:

classquiz/routers/storage.py

@@ -10,7 +10,7 @@ from fastapi.responses import StreamingResponse, RedirectResponse
 from pydantic import BaseModel
 
 from classquiz.auth import get_current_user
-from classquiz.config import settings, storage, arq
+from classquiz.config import settings, storage, arq, ALLOWED_MIME_TYPES
 from classquiz.db.models import User, StorageItem, PublicStorageItem, UpdateStorageItem, PrivateStorageItem
 from classquiz.helpers import check_image_string
 from classquiz.storage.errors import DownloadingFailedError
@@ -119,10 +119,11 @@ async def download_file_head(file_name: str) -> Response:
 
 @router.post("/")
 async def upload_file(file: UploadFile = File(), user: User = Depends(get_current_user)) -> PublicStorageItem:
+    if file.content_type not in ALLOWED_MIME_TYPES:
+        raise HTTPException(status_code=422, detail="Unsupported")
     if user.storage_used > settings.free_storage_limit:
         raise HTTPException(status_code=409, detail="Storage limit reached")
     file_id = uuid4()
-
     size = 0
     file_obj = StorageItem(
         id=file_id,

classquiz/routers/users/__init__.py

@@ -14,7 +14,6 @@ from fastapi import APIRouter, Response, HTTPException, Request, Depends, status
 from fastapi.background import BackgroundTasks
 from fastapi.responses import JSONResponse, RedirectResponse, PlainTextResponse
 from fastapi.security import OAuth2PasswordRequestForm
-import html
 
 from jose import jwt, JWTError
 
@@ -78,7 +77,7 @@ async def create_user(user: RouteUser, background_task: BackgroundTasks) -> User
         raise HTTPException(status_code=409, detail="User already exists")
 
     user.password = get_password_hash(user.password)
-    user.username = html.unescape(bleach.clean(user.username, tags=[], strip=True))
+    user.username = bleach.clean(user.username, tags=[], strip=True)
     if len(user.username) == 32:
         return JSONResponse({"details": "Username mustn't be 32 characters long"}, 400)
     await user.save()

classquiz/routers/users/twofa.py

@@ -7,10 +7,10 @@ import os
 import urllib.parse
 
 import pyotp
-from fastapi import APIRouter, Depends
+from fastapi import APIRouter, Depends, HTTPException
 from pydantic import BaseModel
 
-from classquiz.auth import get_current_user
+from classquiz.auth import get_current_user, verify_password
 from classquiz.db.models import User
 
 router = APIRouter()
@@ -20,10 +20,16 @@ class GetBackupCodeResponse(BaseModel):
     code: str
 
 
-@router.get("/backup_code", response_model=GetBackupCodeResponse)
-async def get_backup_code(user: User = Depends(get_current_user)):
+class RequirePasswordForAction(BaseModel):
+    password: str
+
+
+@router.post("/backup_code", response_model=GetBackupCodeResponse)
+async def get_backup_code(data: RequirePasswordForAction, user: User = Depends(get_current_user)):
     backup_code = os.urandom(32).hex()
     user = await User.objects.get(id=user.id)
+    if not verify_password(data.password, user.password):
+        raise HTTPException(status_code=401, detail="Invalid")
     user.backup_code = backup_code
     await user.update()
     return GetBackupCodeResponse(code=backup_code)
@@ -31,11 +37,14 @@ async def get_backup_code(user: User = Depends(get_current_user)):
 
 class SetRequirePassword(BaseModel):
     require_password: bool
+    password: str
 
 
 @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)
+    if not verify_password(data.password, user.password):
+        raise HTTPException(status_code=401, detail="Invalid")
     user.require_password = data.require_password
     await user.update()
     return data
@@ -47,8 +56,10 @@ class SetTotpUpResponse(BaseModel):
 
 
 @router.post("/totp", response_model=SetTotpUpResponse)
-async def set_totp_up(user: User = Depends(get_current_user)):
+async def set_totp_up(data: RequirePasswordForAction, user: User = Depends(get_current_user)):
     user = await User.objects.get(id=user.id)
+    if not verify_password(data.password, user.password):
+        raise HTTPException(status_code=401, detail="Invalid")
     user.totp_secret = pyotp.random_base32()
     url = pyotp.totp.TOTP(user.totp_secret).provisioning_uri(
         name=urllib.parse.quote(user.username), issuer_name="ClassQuiz"
@@ -71,7 +82,9 @@ async def get_totp_status(user: User = Depends(get_current_user)):
 
 
 @router.delete("/totp")
-async def disable_totp(user: User = Depends(get_current_user)):
+async def disable_totp(data: RequirePasswordForAction, user: User = Depends(get_current_user)):
     user = await User.objects.get(id=user.id)
+    if not verify_password(data.password, user.password):
+        raise HTTPException(status_code=401, detail="Invalid")
     user.totp_secret = None
     await user.update()

classquiz/routers/users/webauthn.py

@@ -8,7 +8,7 @@ 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.auth import get_current_user, verify_password
 from classquiz.db.models import User, FidoCredentials
 from classquiz.config import redis, settings
 
@@ -28,9 +28,15 @@ settings = settings()
 router = APIRouter()
 
 
-@router.get("/add_key", response_model=PublicKeyCredentialCreationOptions)
-async def request_add_key_data(user: User = Depends(get_current_user)):
+class RequirePasswordForAction(BaseModel):
+    password: str
+
+
+@router.post("/add_key_init", response_model=PublicKeyCredentialCreationOptions)
+async def request_add_key_data(data: RequirePasswordForAction, user: User = Depends(get_current_user)):
     user = await User.objects.select_related("fidocredentialss").get(id=user.id)
+    if not verify_password(data.password, user.password):
+        raise HTTPException(status_code=401, detail="Invalid")
     options = generate_registration_options(
         rp_id=urllib.parse.urlparse(settings.root_address).hostname,
         rp_name="ClassQuiz",
@@ -86,7 +92,9 @@ async def list_security_keys(user: User = Depends(get_current_user)):
 
 
 @router.delete("/key/{key_id}")
-async def delete_security_key(key_id: int, user: User = Depends(get_current_user)):
+async def delete_security_key(data: RequirePasswordForAction, key_id: int, user: User = Depends(get_current_user)):
+    if not verify_password(data.password, user.password):
+        raise HTTPException(status_code=401, detail="Invalid")
     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")