119 lines
4.1 KiB
Python
119 lines
4.1 KiB
Python
import os
|
|
import logging
|
|
|
|
from flask import Flask, request, session
|
|
from typing import cast
|
|
from itsdangerous import BadSignature, URLSafeSerializer
|
|
from xkcdpass import xkcd_password
|
|
|
|
from .models import Deck, InfoBanner, Token, db, Course
|
|
|
|
# Create a module-level logger per Flask/Python standards
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def init_password_generator():
|
|
wordfile = xkcd_password.locate_wordfile()
|
|
global words
|
|
words = xkcd_password.generate_wordlist(wordfile=wordfile, min_length=3, max_length=6)
|
|
logger.info("Password generator initialized.")
|
|
|
|
def init_serializer(app: Flask):
|
|
global serializer
|
|
serializer = URLSafeSerializer(app.config['SECRET_KEY'])
|
|
|
|
def get_active_banner() -> InfoBanner | None:
|
|
return db.session.execute(db.select(InfoBanner).where(InfoBanner.is_active)).scalar_one_or_none()
|
|
|
|
def get_available_decks(app: Flask, course: Course):
|
|
slides_dir = os.path.join(app.config['SLIDES_DIR'], course.folder)
|
|
try:
|
|
deck_names = [name for name in os.listdir(slides_dir) if os.path.isdir(os.path.join(slides_dir, name))]
|
|
deck_names.sort()
|
|
return deck_names
|
|
except FileNotFoundError:
|
|
# Log this error as it indicates a configuration or filesystem issue
|
|
logger.error(f"Slides directory not found for course {course.folder}: {slides_dir}")
|
|
return []
|
|
|
|
def generate_username() -> str:
|
|
return cast(str, xkcd_password.generate_xkcdpassword(wordlist=words, numwords=3, delimiter="-"))
|
|
|
|
def generate_password() -> str:
|
|
return cast(str, xkcd_password.generate_xkcdpassword(wordlist=words, numwords=2, delimiter="-"))
|
|
|
|
def get_passwords(course: Course):
|
|
return [password for deck in course.decks for password in deck.passwords]
|
|
|
|
def generate_token() -> Token:
|
|
max_tries = 100
|
|
for _ in range(max_tries):
|
|
user_name = generate_username()
|
|
|
|
if not db.session.execute(db.select(Token).where(Token.name == user_name)).scalar():
|
|
break
|
|
else:
|
|
# Critical error: Application cannot register new users
|
|
logger.critical(f"Failed to generate unique username after {max_tries} tries.")
|
|
raise RuntimeError(f"Unique username could not be generated after {max_tries} tries")
|
|
return create_token(user_name)
|
|
|
|
def create_token(user_name : str) -> Token:
|
|
token = Token(user_name)
|
|
db.session.add(token)
|
|
db.session.commit()
|
|
logger.info(f"Created new token for user: {user_name}")
|
|
return token
|
|
|
|
def serialize_token(token: Token) -> str:
|
|
return serializer.dumps(token.id)
|
|
|
|
def get_token() -> (Token | None):
|
|
token_cookie = request.cookies.get('access_token')
|
|
if not token_cookie:
|
|
return None
|
|
try:
|
|
id = serializer.loads(token_cookie)
|
|
token = db.session.get(Token, id)
|
|
return token
|
|
except BadSignature:
|
|
# Warning: Client sent a tampered or invalid cookie
|
|
logger.warning("Invalid signature for token cookie encountered.")
|
|
return None
|
|
|
|
def retrieve_token(username : str):
|
|
return db.session.execute(db.select(Token).filter_by(name=username)).scalar_one_or_none()
|
|
|
|
def get_unlocked_decks(token: (Token|None)):
|
|
if token is None:
|
|
return []
|
|
decks = [deck for deck in token.decks if deck.course_id == token.active_course_id]
|
|
decks.sort(key=lambda deck: deck.deck)
|
|
return decks
|
|
|
|
def is_deck_unlocked(deck: str):
|
|
token = get_token()
|
|
if token:
|
|
return token.is_unlocked(deck)
|
|
else:
|
|
return False
|
|
|
|
def update_token(decks: list[Deck]):
|
|
"""
|
|
Adds the deck to the list of decks stored in the access token
|
|
"""
|
|
token = get_token()
|
|
if token is None:
|
|
logger.error("Attempted to update token decks, but token is None.")
|
|
raise ValueError("Invalid token is set")
|
|
else:
|
|
# Log the update action
|
|
logger.info(f"Updating decks for user {token.name}. Adding {len(decks)} decks.")
|
|
token.decks = list(set(decks + token.decks))
|
|
db.session.commit()
|
|
|
|
def get_course(id: int) -> (Course | None):
|
|
return db.session.get(Course, id)
|
|
|
|
def get_all_courses() -> list[Course]:
|
|
return db.session.execute(db.select(Course).order_by(Course.id)).scalars().all()
|