224 lines
6.6 KiB
Plaintext
224 lines
6.6 KiB
Plaintext
from flask import render_template, request, make_response, redirect
|
|
import database, re, bcrypt, secrets, time
|
|
|
|
from app import app
|
|
|
|
def get_session(request):
|
|
sessionid = request.cookies.get('session')
|
|
|
|
if not sessionid:
|
|
return False
|
|
|
|
sessionid = str(sessionid)
|
|
sessioninfo = database.fetch("""SELECT uid, created_at, expires, description
|
|
FROM session
|
|
WHERE value=?;""",
|
|
[sessionid])
|
|
|
|
if len(sessioninfo) == 0:
|
|
return False
|
|
|
|
if len(sessioninfo) > 1:
|
|
# there is a problem
|
|
return False
|
|
|
|
return sessioninfo[0] + (sessionid,)
|
|
|
|
def check_pw(uid, password):
|
|
userquery = database.fetch("SELECT uid, password FROM user WHERE username=?", [username])
|
|
|
|
# STOP TIMING ATTACKS
|
|
time.sleep(secrets.randbelow(100)/50)
|
|
|
|
if len(userquery) == 0:
|
|
return render_template("auth/login.html",
|
|
response=f"Der Nutzername >{username} oder das Passwort ist falsch.",
|
|
username=username)
|
|
|
|
uidquery = userquery[0][0]
|
|
passquery = userquery[0][1]
|
|
|
|
passform = str(request.form['pass'])
|
|
passform = passform.encode("utf8")
|
|
passcorrect = bcrypt.checkpw(passform, passquery)
|
|
|
|
if not passcorrect:
|
|
return render_template("auth/login.html",
|
|
response=f"Der Nutzername {username} oder das >Passwort ist falsch.",
|
|
username=username)
|
|
|
|
|
|
@app.route("/auth/headerbar")
|
|
def headerbar():
|
|
sessioninfo = get_session(request)
|
|
|
|
if not sessioninfo:
|
|
return render_template("auth/headerbar.html", loggedin=False)
|
|
|
|
return render_template("auth/headerbar.html", loggedin=True)
|
|
|
|
@app.route("/auth/sessioninfo", methods=['GET'])
|
|
def sessioninfo():
|
|
time.sleep(secrets.randbelow(100)/50) # Rate limiting
|
|
|
|
sessioninfo = get_session(request)
|
|
|
|
if not sessioninfo:
|
|
return "Invalid session"
|
|
|
|
(uid, created_at, expires, description) = sessioninfo
|
|
# uid = sessioninfo[0][0]
|
|
# created_at = sessioninfo[0][1]
|
|
# expires = sessioninfo[0][2]
|
|
# description = sessioninfo[0][3]
|
|
return render_template("/auth/sessioninfo.html",
|
|
sessionid=sessionid,
|
|
uid=uid,
|
|
created_at=created_at,
|
|
expires=expires,
|
|
description=description)
|
|
|
|
@app.route("/auth/logout", methods=['GET'])
|
|
def logout():
|
|
sessioninfo = get_session(request)
|
|
|
|
response = redirect("/")
|
|
response.set_cookie('session', value='', expires=0)
|
|
|
|
if not sessioninfo:
|
|
return response
|
|
|
|
database.execute("DELETE FROM session WHERE value = ?", [sessioninfo[4]])
|
|
|
|
return response
|
|
|
|
@app.route("/auth/login", methods=['GET', 'POST'])
|
|
def login():
|
|
sessioninfo = get_session(request)
|
|
|
|
if sessioninfo:
|
|
uid = sessioninfo[0]
|
|
username = database.fetch("SELECT username FROM user WHERE uid = ?", [uid])
|
|
username = username[0][0]
|
|
return render_template("auth/loginsuccess.html", username=username)
|
|
|
|
|
|
if request.method == 'GET':
|
|
return render_template("auth/login.html",
|
|
response="",
|
|
username="")
|
|
|
|
username = _sanitizeUsername(request.form['user'])
|
|
if len(username) < 3:
|
|
return render_template("auth/login.html",
|
|
response=f"Der Nutzername {username} ist zu kurz.",
|
|
username=username)
|
|
|
|
userquery = database.fetch("SELECT uid, password FROM user WHERE username=?", [username])
|
|
|
|
# STOP TIMING ATTACKS
|
|
time.sleep(secrets.randbelow(100)/50)
|
|
|
|
if len(userquery) == 0:
|
|
return render_template("auth/login.html",
|
|
response=f"Der Nutzername >{username} oder das Passwort ist falsch.",
|
|
username=username)
|
|
|
|
uidquery = userquery[0][0]
|
|
passquery = userquery[0][1]
|
|
|
|
passform = str(request.form['pass'])
|
|
passform = passform.encode("utf8")
|
|
passcorrect = bcrypt.checkpw(passform, passquery)
|
|
|
|
if not passcorrect:
|
|
return render_template("auth/login.html",
|
|
response=f"Der Nutzername {username} oder das >Passwort ist falsch.",
|
|
username=username)
|
|
|
|
now = int(time.time())
|
|
|
|
timeout = int(request.form['timeout'])
|
|
timeout = now + timeout
|
|
|
|
session = secrets.randbelow(999999999)
|
|
session = str(session) + str(now)
|
|
|
|
database.execute("DELETE FROM session WHERE expires < ?", [str(now)])
|
|
database.execute("""INSERT INTO session
|
|
(uid, created_at, expires, description, value)
|
|
VALUES (?, ?, ?, ?, ?);""",
|
|
(uidquery, int(time.time()), timeout, "", session)
|
|
)
|
|
|
|
response = make_response("<script>location.href = '/';</script>")
|
|
response.set_cookie('session', value=session, expires=timeout)
|
|
return response
|
|
|
|
@app.route("/auth/newuser", methods=['GET', 'POST'])
|
|
def newuser():
|
|
if request.method == 'GET':
|
|
return render_template("auth/register.html",
|
|
responses=[],
|
|
username="")
|
|
|
|
success = True
|
|
responses = []
|
|
|
|
username = _sanitizeUsername(request.form['user'])
|
|
if len(username) < 3:
|
|
success = False
|
|
responses.append(f"Der Nutzername {username} ist zu kurz.")
|
|
usernamequery = database.fetch("SELECT id FROM user WHERE username=?", [username])
|
|
if len(usernamequery) != 0:
|
|
success = False
|
|
responses.append(f"Der Nutzername {username} wird bereits verwendet.")
|
|
|
|
pass1 = str(request.form['pass1'])
|
|
pass2 = str(request.form['pass2'])
|
|
if pass1 != pass2:
|
|
success = False
|
|
responses.append(f"Passwort und Passwortbestätigung sind ungleich.")
|
|
|
|
if len(pass1) < 10:
|
|
success = False
|
|
responses.append(f"Das Passwort ist kürzer als 10 Zeichen.")
|
|
|
|
if not success:
|
|
return render_template("auth/register.html",
|
|
responses=responses,
|
|
username=username)
|
|
|
|
uid = secrets.randbelow(99999999)
|
|
created_at = int(time.time())
|
|
# emailcode = secrets.randbelow(999999)
|
|
passsalt = bcrypt.gensalt()
|
|
passhash = bcrypt.hashpw(pass1.encode("utf8"), passsalt)
|
|
|
|
database.execute("""INSERT INTO user
|
|
(uid, created_at, username, password) VALUES
|
|
(?, ?, ?, ?)""",
|
|
(uid, created_at, username, passhash))
|
|
|
|
return render_template("auth/registersuccess.html", username=username)
|
|
|
|
def _sanitizeUsername(username):
|
|
username = str(username)
|
|
username = username.strip()
|
|
username = username[:32]
|
|
|
|
# Remove all non-word characters (everything except numbers and letters)
|
|
username = re.sub(r"[^._\w\s-]", '', username)
|
|
|
|
# Replace all runs of whitespace with a single dash
|
|
username = re.sub(r"\s+", '_', username)
|
|
|
|
return username
|
|
|
|
def _verifyEmail(email):
|
|
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
|
|
if re.match(pattern, email):
|
|
return True
|
|
else:
|
|
return False
|