PatchBay/web/patchbay_web/auth.py

"""Login (password + emailed one-time code), sessions and CSRF protection."""

import functools
import hmac
import secrets
import time

from flask import (Blueprint, abort, current_app, flash, g, redirect, render_template, request,
                   session, url_for)

from . import db, mailer, security

bp = Blueprint("auth", __name__)

CODE_TTL = 600
CODE_ATTEMPTS = 5
FAIL_WINDOW = 900
FAIL_LIMIT = 10


def conf():
    return current_app.config["PB_CONF"]


def lookup_user(username):
    """Returns dict(username, email, pwhash, sysop) or None."""
    c = conf()
    if username == c.get("sysopuser"):
        return {"username": username, "email": c.get("sysopemail"), "pwhash": c.get("sysoppassword"),
                "sysop": True}
    row = db.get().execute("SELECT username, email, pwhash FROM users WHERE username = ?", (username,)).fetchone()
    if row:
        return {"username": row["username"], "email": row["email"], "pwhash": row["pwhash"], "sysop": False}
    return None


def _too_many_failures(addr):
    cutoff = db.now() - FAIL_WINDOW
    conn = db.get()
    conn.execute("DELETE FROM login_failures WHERE ts < ?", (cutoff,))
    n = conn.execute("SELECT COUNT(*) FROM login_failures WHERE addr = ?", (addr,)).fetchone()[0]
    return n >= FAIL_LIMIT


def _record_failure(addr):
    db.get().execute("INSERT INTO login_failures (addr, ts) VALUES (?, ?)", (addr, db.now()))


def csrf_token():
    if "csrf" not in session:
        session["csrf"] = secrets.token_urlsafe(32)
    return session["csrf"]


@bp.before_app_request
def check_csrf():
    if request.method in ("GET", "HEAD", "OPTIONS"):
        return
    sent = request.headers.get("X-CSRF-Token") or request.form.get("csrf_token", "")
    if not sent or not hmac.compare_digest(sent, session.get("csrf", "")):
        abort(400, "invalid CSRF token")


@bp.before_app_request
def load_user():
    g.user = None
    name = session.get("user")
    if not name:
        return
    if time.time() - session.get("login_at", 0) > conf().getint("sessionhours", 12) * 3600:
        session.clear()
        return
    user = lookup_user(name)
    # Password changes invalidate other sessions of that user.
    if user and session.get("pwtag") == user["pwhash"][-16:]:
        g.user = user
    else:
        session.clear()


def login_required(view):
    @functools.wraps(view)
    def wrapped(*a, **kw):
        if g.user is None:
            if request.path.startswith("/api/"):
                abort(401)
            return redirect(url_for("auth.login"))
        return view(*a, **kw)
    return wrapped


def sysop_required(view):
    @functools.wraps(view)
    @login_required
    def wrapped(*a, **kw):
        if not g.user["sysop"]:
            abort(403)
        return view(*a, **kw)
    return wrapped


@bp.route("/login", methods=["GET", "POST"])
def login():
    if g.user:
        return redirect(url_for("views.patch"))
    if request.method == "GET":
        return render_template("login.html")

    addr = request.remote_addr or "?"
    if _too_many_failures(addr):
        flash("Too many failed attempts. Try again later.", "error")
        return render_template("login.html"), 429

    username = request.form.get("username", "").strip()
    password = request.form.get("password", "")
    user = lookup_user(username)
    if not security.verify_password(password, user["pwhash"] if user else None) or not user:
        _record_failure(addr)
        flash("Invalid username or password.", "error")
        return render_template("login.html"), 401
    if not user["email"]:
        flash("This account has no email address for the login code.", "error")
        return render_template("login.html"), 400

    tok, code = security.token(), security.new_code()
    conn = db.get()
    conn.execute("DELETE FROM login_codes WHERE expires < ? OR username = ?", (db.now(), username))
    conn.execute("INSERT INTO login_codes (token, username, codehash, expires) VALUES (?, ?, ?, ?)",
                 (tok, username, security.code_hash(tok, code), db.now() + CODE_TTL))
    try:
        mailer.send(conf(), user["email"], "PatchBay login code",
                    f"Your PatchBay login code is: {code}\n\n"
                    f"It is valid for {CODE_TTL // 60} minutes. If you did not try to log in, "
                    f"someone knows your password; change it.\n")
    except mailer.MailError as e:
        current_app.logger.error("sending login code failed: %s", e)
        flash("Could not send the login code. Check the mail settings.", "error")
        return render_template("login.html"), 503

    session.clear()
    session["pending"] = tok
    return redirect(url_for("auth.verify"))


@bp.route("/verify", methods=["GET", "POST"])
def verify():
    tok = session.get("pending")
    if not tok:
        return redirect(url_for("auth.login"))
    if request.method == "GET":
        return render_template("verify.html")

    conn = db.get()
    row = conn.execute("SELECT username, codehash, expires, attempts FROM login_codes WHERE token = ?",
                       (tok,)).fetchone()
    if not row or row["expires"] < db.now() or row["attempts"] >= CODE_ATTEMPTS:
        conn.execute("DELETE FROM login_codes WHERE token = ?", (tok,))
        session.clear()
        flash("The login code expired. Please log in again.", "error")
        return redirect(url_for("auth.login"))

    code = request.form.get("code", "").strip()
    if not hmac.compare_digest(security.code_hash(tok, code), row["codehash"]):
        conn.execute("UPDATE login_codes SET attempts = attempts + 1 WHERE token = ?", (tok,))
        _record_failure(request.remote_addr or "?")
        flash("Wrong code.", "error")
        return render_template("verify.html"), 401

    conn.execute("DELETE FROM login_codes WHERE token = ?", (tok,))
    user = lookup_user(row["username"])
    if not user:
        session.clear()
        return redirect(url_for("auth.login"))
    session.clear()
    session.permanent = True
    session["user"] = user["username"]
    session["login_at"] = time.time()
    session["pwtag"] = user["pwhash"][-16:]
    csrf_token()
    return redirect(url_for("views.patch"))


@bp.route("/logout", methods=["POST"])
def logout():
    session.clear()
    return redirect(url_for("auth.login"))


def refresh_pwtag():
    """Keeps the current session valid after the user changed their own password."""
    user = lookup_user(session["user"])
    session["pwtag"] = user["pwhash"][-16:]