From 61a0a0a6a6da081e1a43093789a6b909ebbd96b7 Mon Sep 17 00:00:00 2001 From: Christian Krakau-Louis Date: Wed, 16 Apr 2025 11:29:52 +0200 Subject: [PATCH] feat: Implement initial setup functionality with admin elevation and system settings management --- app/db_init.py | 21 ++- app/main.py | 3 +- app/models.py | 12 ++ app/templates/admin/setup.html | 226 +++++++++++++++++++++++++++++++++ app/views/setup.py | 128 +++++++++++++++++++ 5 files changed, 388 insertions(+), 2 deletions(-) create mode 100644 app/templates/admin/setup.html create mode 100644 app/views/setup.py diff --git a/app/db_init.py b/app/db_init.py index b42927f..c7bbc39 100644 --- a/app/db_init.py +++ b/app/db_init.py @@ -9,7 +9,7 @@ from sqlalchemy import inspect from sqlalchemy.orm import Session from passlib.context import CryptContext -from .models import User, Team, TeamMembership, QRCode, QRSet, TeamAchievement, Event +from .models import User, Team, TeamMembership, QRCode, QRSet, TeamAchievement, Event, SystemSettings from .db import SessionLocal, engine from .db_migrations import run_migrations @@ -35,6 +35,25 @@ def init_db(): run_migrations(engine) # Then proceed with seeding if needed seed_db() + # Initialize system settings if needed + init_system_settings() + +def init_system_settings(): + """Initialize the system settings table if it doesn't exist.""" + db = SessionLocal() + try: + # Check if there's already a system settings record + settings = db.query(SystemSettings).first() + if not settings: + # Create initial system settings with setup_completed = False + settings = SystemSettings(setup_completed=False) + db.add(settings) + db.commit() + print("System settings initialized.") + except Exception as e: + print(f"Error initializing system settings: {e}") + finally: + db.close() def seed_db(): """Seed the database with test data.""" diff --git a/app/main.py b/app/main.py index 61b9ca7..c0af7a3 100644 --- a/app/main.py +++ b/app/main.py @@ -15,7 +15,7 @@ import contextlib from .db import engine, get_db from . import models from .templates_config import templates -from .views import qr, redeem, teams, admin, leaderboard, dashboard, static, pages, auth, convenience +from .views import qr, redeem, teams, admin, leaderboard, dashboard, static, pages, auth, convenience, setup from .db_init import init_db from .auth.middleware import SessionAuthBackend, on_auth_error @@ -125,6 +125,7 @@ async def not_found_exception_handler(request: Request, exc): ) # Routers +app.include_router(setup.router, tags=["Setup"]) # Setup router for initial admin setup app.include_router(pages.router, tags=["Pages"]) # Pages router for index and static pages app.include_router(auth.router, prefix="/auth", tags=["auth"]) # Include the auth router app.include_router(qr.router, prefix="/qr", tags=["QR"]) diff --git a/app/models.py b/app/models.py index 892a90f..79ab5de 100644 --- a/app/models.py +++ b/app/models.py @@ -66,6 +66,18 @@ class User(Base, BaseUser): return f"" +# Add SystemSettings model for storing setup configuration +class SystemSettings(Base): + __tablename__ = "system_settings" + id = Column(Integer, primary_key=True, autoincrement=True) + setup_completed = Column(Boolean, default=False, nullable=False) + created_at = Column(DateTime, server_default=func.now()) + updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now()) + + def __repr__(self): + return f"" + + class OAuthAccount(Base): __tablename__ = "oauth_accounts" id = Column(Integer, primary_key=True, index=True) diff --git a/app/templates/admin/setup.html b/app/templates/admin/setup.html new file mode 100644 index 0000000..18bc457 --- /dev/null +++ b/app/templates/admin/setup.html @@ -0,0 +1,226 @@ +{% extends "base.html" %} + +{% block title %}LeagueLedger Setup{% endblock %} + +{% block content %} +
+
+
+
+ +
+

+ LeagueLedger Setup +

+
+ + +
+
+
+ +

Welcome to the LeagueLedger setup page. This page is only available once during initial setup.

+
+
+ + {% if setup_completed %} +
+
+ +

Setup has been completed. The system has been configured with an administrator.

+
+
+ + {% if is_admin %} +

You already have administrator privileges. You can:

+
+ + Go to Admin Dashboard + +
+ {% else %} +

This setup has already been completed by another user. Contact an administrator if you need admin access.

+ {% endif %} + {% else %} +

Welcome, {{ user.username }}!

+ +
+

System Setup

+

+ You're about to be promoted to administrator. As an administrator, you will be able to: +

+
    +
  • Access all administrative functions
    • +
  • Manage users and teams
    • +
  • Generate and manage QR codes
    • +
  • Configure system settings
    • +
+
+ +
+ +
+ {% endif %} +
+ + +
+
+ + Return to Home + + + {% if not setup_completed %} + + + This page will be disabled after setup + + {% endif %} +
+
+
+
+
+
+ + + + + + +{% endblock %} + +{% block extra_scripts %} + +{% endblock %} \ No newline at end of file diff --git a/app/views/setup.py b/app/views/setup.py new file mode 100644 index 0000000..41a3b35 --- /dev/null +++ b/app/views/setup.py @@ -0,0 +1,128 @@ +#!/usr/bin/env python3 +from fastapi import APIRouter, Request, Depends, Form, HTTPException, status +from fastapi.responses import HTMLResponse, RedirectResponse, JSONResponse +from fastapi.templating import Jinja2Templates +from typing import Optional +from sqlalchemy.orm import Session +from starlette.status import HTTP_303_SEE_OTHER, HTTP_401_UNAUTHORIZED +from datetime import datetime +import logging + +# Set up logging +logger = logging.getLogger(__name__) + +from ..db import get_db +from ..models import User, SystemSettings +from ..templates_config import templates +from ..security import verify_password +from ..auth.oauth import oauth_manager + +# Create router +router = APIRouter(tags=["Setup"]) + +@router.get("/setup", response_class=HTMLResponse) +async def setup_page(request: Request, db: Session = Depends(get_db)): + """ + Setup page that allows a logged-in user to elevate themselves to admin. + Only available once before being deactivated. + """ + # Check if user is logged in + user_id = request.session.get("user_id") + if not user_id: + logger.warning("Unauthorized access attempt to setup page") + # Add a query parameter for redirect back to setup after login + return RedirectResponse("/auth/login?next=/setup", status_code=HTTP_303_SEE_OTHER) + + # Get the logged-in user + user = db.query(User).filter(User.id == user_id).first() + if not user: + logger.warning(f"User ID {user_id} found in session but not in database") + request.session.clear() + return RedirectResponse("/auth/login?next=/setup", status_code=HTTP_303_SEE_OTHER) + + # Check if setup has already been completed + settings = db.query(SystemSettings).first() + setup_completed = settings and settings.setup_completed + + # Check if the current user is already admin + is_admin = user.is_admin + + return templates.TemplateResponse( + "admin/setup.html", + { + "request": request, + "user": user, + "setup_completed": setup_completed, + "is_admin": is_admin + } + ) + +@router.post("/setup/elevate", response_class=JSONResponse) +async def elevate_to_admin(request: Request, db: Session = Depends(get_db)): + """ + API endpoint to elevate current user to admin and mark setup as completed. + Returns JSON response for AJAX handling. + """ + # Check if user is logged in + user_id = request.session.get("user_id") + if not user_id: + logger.warning("Unauthorized API call to elevate_to_admin") + return JSONResponse( + status_code=HTTP_401_UNAUTHORIZED, + content={"success": False, "message": "Authentication required"} + ) + + # Check if setup has already been completed + settings = db.query(SystemSettings).first() + if settings and settings.setup_completed: + logger.warning("Setup already completed, but elevate_to_admin was called") + return JSONResponse( + content={ + "success": False, + "message": "Setup has already been completed" + } + ) + + try: + # Get the logged-in user + user = db.query(User).filter(User.id == user_id).first() + if not user: + logger.warning(f"User ID {user_id} not found in database during elevate_to_admin") + return JSONResponse( + status_code=HTTP_401_UNAUTHORIZED, + content={"success": False, "message": "User not found"} + ) + + # Elevate user to admin + user.is_admin = True + user.last_login = datetime.utcnow() + + # Mark setup as completed in system settings + if not settings: + settings = SystemSettings(setup_completed=True) + db.add(settings) + else: + settings.setup_completed = True + + # Save the changes with explicit commit + db.commit() + logger.info(f"User {user.username} (ID: {user.id}) elevated to admin in setup") + + # Update the session to reflect admin status + request.session["is_admin"] = True + + return JSONResponse( + content={ + "success": True, + "message": "You have been successfully promoted to administrator" + } + ) + except Exception as e: + db.rollback() + logger.error(f"Error during admin elevation: {str(e)}") + return JSONResponse( + content={ + "success": False, + "message": f"An error occurred: {str(e)}" + } + ) \ No newline at end of file