app.py

import os
from dotenv import load_dotenv
# Imports
from flask import Flask, render_template, request, redirect, url_for, flash, get_flashed_messages, Markup
from flask_sqlalchemy import SQLAlchemy
from flask_login import LoginManager, login_user, logout_user, login_required, current_user
from flask_talisman import Talisman
from flask_limiter import Limiter
from flask_limiter.util import get_remote_address
from markdown import markdown
# Custom imports
from init_db import initialise_db
from helpers import custom_logger, add_pageview
from models.Comment import Comment
# Models
from models.User import User
from models.Course import Course
# Forms
from forms import LoginForm, RegistrationForm, ModuleForm, CommentForm

load_dotenv()

db = SQLAlchemy()

app = Flask(__name__)
app.config.update(
    SECRET_KEY=os.getenv('SECRET_KEY'),
    SQLALCHEMY_DATABASE_URI=os.getenv('SQLALCHEMY_DATABASE_URI'),
)

SELF = "'self'"

# https://github.com/wntrblm/flask-talisman
# can still add more sources to the content_security_policy for things like bootstrap and jquery
talisman = Talisman(
    app,
    content_security_policy={
        'default-src': SELF,
        'script-src': SELF,
        'object-src': SELF,
        'frame-ancestors': SELF,
        'form-action': SELF,
    },
    strict_transport_security=False,
    session_cookie_secure=False,
    session_cookie_http_only=True,
    session_cookie_samesite='Strict',
    x_content_type_options=True,
    x_xss_protection=True,
    frame_options='DENY',
    force_https=False,
)

# strict_transport_security, session_cookie_secure and force_https are intentionally set to False for development purposes
# this of course allows for insecure transport - https://owasp.org/www-community/vulnerabilities/Insecure_Transport

db.init_app(app)  # not to be confused with init_db.py


@app.after_request
def after_request(response):
    if current_user.is_authenticated:
        add_pageview(db, current_user, request, response)
    return response


@app.template_filter('markdown')
def md(s):
    markd = markdown(s)
    safe_md = Markup(markd)
    return safe_md


limiter = Limiter(
    get_remote_address,
    app=app,
    default_limits=["240 per day", "20 per hour"],
    storage_uri='memory://'
)

logger = custom_logger('app', 'app.log')

login_manager = LoginManager()
login_manager.init_app(app)
login_manager.login_view = 'login'


@login_manager.user_loader
def load_user(user_id):
    return db.session.query(User).get(int(user_id))


@app.route('/login', methods=['GET', 'POST'])
@limiter.limit("5 per hour")
def login():
    form = LoginForm()

    if form.validate_on_submit():
        email = form.email.data.lower()
        password = form.password.data
        user = db.session.query(User).filter_by(email=email).first()
        if user and user.verify_password(password):
            login_user(user)
            logger.info(f'User with ID {user.id} logged in')
            return redirect(url_for('index'))
        else:
            flash('Invalid username or password', 'error')

    return render_template('login.html', form=form)


@app.route('/logout')
def logout():
    logout_user()
    return redirect(url_for('index'))


@app.route('/register', methods=['GET', 'POST'])
@limiter.limit("5 per minute")
def register():
    form = RegistrationForm()
    if form.validate_on_submit():
        if form.errors:
            logger.error(f'Form errors: {form.errors}')
            flash('Form errors', 'error')
            return redirect(url_for('register'))
        email = form.email.data.lower()
        password = form.password.data

        user = db.session.query(User).filter_by(email=email).first()
        if user:
            flash('A user with that email already exists', 'error')
            return redirect(url_for('register'))

        new_user = User(email=email)
        new_user.set_password(password)
        try:
            db.session.add(new_user)
            db.session.commit()
        except Exception as e:
            logger.error(f'Error creating user: {e}')
            flash('Error creating user', 'error')
            return redirect(url_for('register'))

        logger.info(f'User {email} created successfully')
        flash('Registered successfully', 'success')

    return render_template('register.html', form=form)


@app.route('/')
def index():
    return render_template('index.html', user=current_user)


@app.route('/mycourses')
@login_required
def courses():
    return render_template('courses.html', user=current_user, form=ModuleForm())


@app.route('/mycourses/add', methods=['GET', 'POST'])
@login_required
def add_course():
    form = ModuleForm()
    if form.validate_on_submit():
        module_code = form.code.data
        module_title = form.name.data
        module_description = form.description.data
        module = db.session.query(Course).filter_by(code=module_code).first()
        if module:
            flash('Module already exists', 'error')
            return redirect(url_for('courses'))
        else:
            new_module = Course(code=module_code, name=module_title, description=module_description)
            new_module.add_to_db(db)
            flash('Module added successfully', 'success')
            return redirect(url_for('courses'))
    return render_template('courses.html', form=form, user=current_user)


@app.route('/mycourses/<course_id>/', methods=['GET', 'POST'])
@login_required
def course_info(course_id):
    form = CommentForm()
    course = db.session.query(Course).filter_by(id=course_id).first()
    if course:

        if not course.is_enrolled(current_user):
            if course.is_taught_by(current_user):
                pass
            else:
                flash(f'You are not enrolled in this course: {course.code}', 'error')
                return redirect(url_for('courses'))

        if form.validate_on_submit():
            comment = form.text.data
            new_comment = Comment(user_id=current_user.id, course_id=course_id, user=current_user, course=course,
                                  text=comment)
            new_comment.add_to_db(db.session)
            logger.info(f'User {current_user.id} added comment to course {course_id}')
            # loqger.info(f'User {current_user.id} added comment to course {course_id}')
            flash('Comment added successfully', 'success')
            return redirect(url_for('course_info', course_id=course_id))
        return render_template('course_info.html', course=course, user=current_user, form=form,
                               comments=course.comments)
    else:
        flash('Course not found', 'error')
        return redirect(url_for('courses'))


@app.route('/admin_area')
@login_required
def admin_area():
    if current_user.role != 'admin':
        logger.warn(f'User with ID {current_user.id} attempted to access admin area (admin_area).')
        flash('You do not have permission to access this area: Admin Area', 'error')
        return redirect(url_for('index'))
    return render_template('admin_area.html', user=current_user)


@app.route('/initdb')
def init_db_route():
    logger.info("Running database initialisation code")
    # debug only route to initialise the database
    if initialise_db(db.engine):
        logger.info("Database initialised successfully")
        return 'Database initialised'
    else:
        logger.error("Database initialisation failed")
        return 'Database initialisation failed. Check the logs for more information'


@app.route('/debug_users')
@login_required
def debug_users():
    if current_user.role != 'admin':
        logger.warn(f'User with ID {current_user.id} attempted to access admin area (debug_users).')
        flash('You do not have permission to access this area: Debug Users', 'error')
        return redirect(url_for('index'))
    users = db.session.query(User).all()
    return render_template('debug_users.html', users=users, user=current_user)


@app.route('/debug_courses')
@login_required
def debug_courses():
    if current_user.role != 'admin':
        logger.warn(f'User with ID {current_user.id} attempted to access admin area (debug_courses).')
        flash('You do not have permission to access this area: Debug Courses', 'error')
        return redirect(url_for('index'))
    courses = db.session.query(Course).all()
    return render_template('debug_courses.html', courses=courses, user=current_user)


if __name__ == '__main__':
    if os.path.exists('app.db') or os.path.exists('instance/app.db'):
        app.run()
    else:
        print('Database not initialised. Run init_db.py or access /initdb')