pyrom/app/routes/users.py

from flask import Blueprint, redirect, url_for, render_template, request, session, abort
from functools import wraps
import time

from ..auth import (
    digest, verify, create_session,
    is_logged_in, parse_username, is_password_valid,
    login_required, revoke_session, get_active_user
    )
from ..models import Users, Posts, Reactions, Threads
from ..constants import PermissionLevel
from secrets import compare_digest as compare_timesafe
import math

bp = Blueprint('users', __name__, url_prefix='/users/')

def redirect_if_logged_in(destination='topics.all_topics'):
    def decorator(view_func):
        @wraps(view_func)
        def wrapper(*args, **kwargs):
            if is_logged_in():
                return redirect(url_for(destination))
            return view_func(*args, **kwargs)
        return wrapper
    return decorator

def redirect_to_own(view_func):
    @wraps(view_func)
    def wrapper(username, *args, **kwargs):
        user = get_active_user()
        if username.lower() != user.username:
            view_args = dict(request.view_args)
            view_args.pop('username', None)
            new_args = {**view_args, 'username': user.username}
            return redirect(url_for(request.endpoint, **new_args))
        return view_func(username, *args, **kwargs)
    return wrapper


@bp.get('/log-in/')
@redirect_if_logged_in()
def log_in():
    return render_template('users/log_in.html')

@bp.post('/log-in/')
@redirect_if_logged_in()
def log_in_post():
    username = request.form.get('username', default='').lower()
    user = Users.find({'username': username})
    if not user:
        return redirect(url_for('.log_in', error='The username or password you entered is incorrect.'))
    password = request.form.get('password', default='')
    if not verify(user.password_hash, password):
        return redirect(url_for('.log_in', error='The username or password you entered is incorrect.'))

    session['remember'] = request.form.get('remember') == 'on'
    sess = create_session(user.id, not session['remember'])
    session['pyrom_session_key'] = sess.key
    if session['remember']:
        session.permanent = True
    return redirect(request.form.get('return_to', default=url_for('topics.all_topics')))

@bp.post('/log-out/')
@login_required
def log_out():
    revoke_session(get_active_user().id)
    return redirect(url_for('topics.all_topics'))

@bp.get('/sign-up/')
@redirect_if_logged_in()
def sign_up():
    return render_template('users/sign_up.html')

@bp.post('/sign-up/')
@redirect_if_logged_in()
def sign_up_post():
    generic_error_page = redirect(url_for('.sign_up', error='The username or password you entered is invalid.'))
    invalid_username_error_page = redirect(url_for('.sign_up', error='This username cannot be used. Please pick another.'))
    passwords_error_page = redirect(url_for('.sign_up', error='The passwords do not match.'))
    username = request.form.get('username', default='')
    if not username:
        return generic_error_page
    if request.form.get('password') is None:
        return generic_error_page
    if len(request.form.getlist('password')) != 2:
        return passwords_error_page
    try:
        username_pair = parse_username(username)
    except ValueError:
        return invalid_username_error_page
    potential_user = Users.find({'username': username})
    if potential_user:
        return invalid_username_error_page

    if not compare_timesafe(request.form.getlist('password')[0], request.form.getlist('password')[1]):
        return passwords_error_page

    password_hash = digest(request.form.get('password'))

    user = Users.create({
        'username': username_pair[0],
        'password_hash': password_hash,
        'permission': PermissionLevel.GUEST.value,
        'created_at': int(time.time()),
    })

    if username_pair[0] != username_pair[1]:
        user.update({
            'display_name': username_pair[1]
        })

    session['remember'] = request.form.get('remember') == 'on'
    sess = create_session(user.id, not session['remember'])
    session['pyrom_session_key'] = sess.key
    if session['remember']:
        session.permanent = True

    return redirect(url_for('topics.all_topics'))

@bp.get('/<username>/')
def user_page(username):
    username = username.lower()
    target_user = Users.find({'username': username})
    if not target_user:
        abort(404)
    return render_template('users/user_page.html', target_user=target_user)

@bp.get('/<username>/posts/')
def posts(username):
    username = username.lower()
    if username == 'deleteduser':
        abort(404)
    target_user = Users.find({'username': username})
    if not target_user:
        abort(404)
    PER_PAGE = 10
    posts_count = Posts.count({'user_id': target_user.id})
    page_count = max(1, math.ceil(posts_count / PER_PAGE))
    page = 1
    try:
        page = max(1, min(int(request.args.get('page', default=1)), page_count))
    except ValueError:
        abort(404)
    posts = target_user.get_posts(PER_PAGE, page)
    return render_template(
        'users/posts.html', posts=posts,
        page=page, page_count=page_count,
        target_user=target_user, Reactions=Reactions
        )

@bp.get('/<username>/threads/')
def threads(username):
    username = username.lower()
    if username == 'deleteduser':
        abort(404)
    target_user = Users.find({'username': username})
    if not target_user:
        abort(404)
    PER_PAGE = 10
    threads_count = Threads.count({'user_id': target_user.id})
    page_count = max(1, math.ceil(threads_count / PER_PAGE))
    page = 1
    try:
        page = max(1, min(int(request.args.get('page', default=1)), page_count))
    except ValueError:
        abort(404)
    threads = target_user.get_started_threads(PER_PAGE, page)
    return render_template(
        'users/threads.html', threads=threads,
        page=page, page_count=page_count,
        target_user=target_user, Reactions=Reactions
        )

@bp.get('/<username>/comments/')
def comments(username):
    username = username.lower()
    if username == 'deleteduser':
        abort(404)
    return 'stub'

@bp.get('/<username>/settings/')
@login_required
@redirect_to_own
def settings(username):
    username = username.lower()
    return 'stub'

@bp.get('/<username>/inbox/')
@login_required
@redirect_to_own
def inbox(username):
    user = get_active_user()
    unread_count = user.get_unread_count()
    subscriptions = user.get_all_subscriptions()
    return render_template('users/inbox.html', unread_count=unread_count, subscriptions=subscriptions)

@bp.get('/<username>/bookmarks/')
@login_required
@redirect_to_own
def bookmarks(username):
    username = username.lower()
    return 'stub'