import os import logging from flask import request, session, redirect, url_for, current_app from typing import cast from itsdangerous import BadSignature, URLSafeSerializer from xkcdpass import xkcd_password from functools import wraps, lru_cache 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): 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(course: Course): slides_dir = os.path.join(current_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 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() def token_required(f): @wraps(f) def decorated_function(*args, **kwargs): token = get_token() if not token: logger.debug("User has no token. Redirecting to set_token.") # --- Capture the target URL --- # request.url captures the full URL including query parameters # (e.g., /dashboard?section=homework) session["next_url"] = request.url return redirect(url_for("main.set_token")) kwargs["token"] = token return f(*args, **kwargs) return decorated_function # Cache the last 1000 lookups per worker. @lru_cache(maxsize=1000) def get_cached_deck_info(token_id, course_folder, deck_slug): """ Returns a dictionary of primitive types for route authorization and routing. Never return raw SQLAlchemy models from an lru_cache! """ # 1. Fetch Token token = db.session.get(Token, token_id) if not token: return {"auth": False, "course_exists": False, "deck_exists": False} # 2. Fetch Course course = db.session.execute( db.select(Course).where(Course.folder == course_folder) ).scalar() if not course: return {"auth": False, "course_exists": False, "deck_exists": False} # 3. Fetch Deck deck_obj = db.session.execute( db.select(Deck).where(Deck.deck == deck_slug, Deck.course_id == course.id) ).scalar() if not deck_obj: return {"auth": False, "course_exists": True, "deck_exists": False} # 4. Check Authorization is_unlocked = deck_slug in [d.deck for d in token.decks] # Return only primitive data (bools, strings) so the cache is thread-safe! return { "auth": is_unlocked, "course_exists": True, "deck_exists": True, "deck_name": deck_obj.name, "index_file": deck_obj.index_file }