241 lines
7.0 KiB
Python
241 lines
7.0 KiB
Python
import os
|
|
import logging
|
|
|
|
from flask import request, session, redirect, url_for, current_app, abort, make_response
|
|
from typing import cast
|
|
from itsdangerous import BadSignature, URLSafeSerializer
|
|
from xkcdpass import xkcd_password
|
|
from functools import wraps, lru_cache
|
|
from urllib.parse import urlparse, urljoin
|
|
|
|
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 set_cookie(token):
|
|
next_url = session.pop("next_url", None)
|
|
if not next_url:
|
|
current_app.logger.debug("next_url not present redirecting to index")
|
|
next_url = url_for("main.index")
|
|
|
|
if not is_safe_url(next_url):
|
|
current_app.logger.warning(f"Blocked unsafe redirect attempt to: {next_url}")
|
|
return abort(400)
|
|
|
|
resp = make_response(redirect(next_url))
|
|
resp.set_cookie(
|
|
"access_token", serialize_token(token), httponly=True, samesite="Lax", path="/", max_age=31536000
|
|
)
|
|
return resp
|
|
|
|
|
|
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
|
|
}
|
|
|
|
def is_safe_url(target):
|
|
"""
|
|
Ensures that a redirect target will lead to the same server.
|
|
Handles both relative paths ('/admin') and absolute URLs ('https://yourdomain.com/admin').
|
|
"""
|
|
ref_url = urlparse(request.host_url)
|
|
test_url = urlparse(urljoin(request.host_url, target))
|
|
|
|
return test_url.scheme in ('http', 'https') and ref_url.netloc == test_url.netloc
|