1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
|
from flask import Flask, request, render_template, redirect, make_response
import psycopg2
import argon2
import tomllib as toml
import secrets
from dataclasses import dataclass
from threading import Timer
from time import time
@dataclass
class User:
username: str
expire: int
timer: Timer
def createUserSession(username: str) -> str:
session_id = secrets.token_urlsafe(16)
verifiedUsers.add(username)
verifiedSessions[session_id] = User(
username, time() + app.config["SESSION_TIMEOUT"], None
)
timer = Timer(app.config["SESSION_TIMEOUT"], expireSession, args=[session_id])
timer.start()
verifiedSessions[session_id].timer = timer
return session_id
def expireSession(session_id: str):
if not verifiedSessions.get(session_id):
return
userSession: User = verifiedSessions[session_id]
userSession.timer.cancel()
verifiedUsers.remove(userSession.username)
del verifiedSessions[session_id]
def configurationReader():
with open("config.toml", "r") as config_file:
config = toml.load(config_file)
app.config["DB_HOST"] = config["database"]["host"]
app.config["DB_USER"] = config["database"]["user"]
app.config["DB_PASSWORD"] = config["database"]["password"]
app.config["DB_NAME"] = config["database"]["name"]
app.config["DB_PORT"] = config["database"]["port"]
app.config["SESSION_TIMEOUT"] = config["session"]["timeout"]
verifiedUsers = set()
verifiedSessions = dict()
app = Flask(__name__)
configurationReader()
@app.route("/")
def index():
if request.cookies.get("lnsession"):
session_id = request.cookies.get("lnsession")
if verifiedSessions.get(session_id):
return render_template("index.html", logged_in=True)
return render_template("index.html", logged_in=False)
@app.route("/login", methods=["POST"])
def login():
page = request.args.get("page")
username = request.form.get("username")
password = request.form.get("password")
psycopg2_connection = psycopg2.connect(
user=app.config["DB_USER"],
password=app.config["DB_PASSWORD"],
host=app.config["DB_HOST"],
port=app.config["DB_PORT"],
database=app.config["DB_NAME"],
)
cur = psycopg2_connection.cursor()
cur.execute("SELECT password FROM users WHERE username = ?", (username,))
result = cur.fetchone()
cur.close()
psycopg2_connection.close()
if result is None:
return render_template("login.html", error="Invalid username or password"), 401
stored_password = result[0]
ph = argon2.PasswordHasher()
try:
ph.verify(stored_password, password)
except argon2.exceptions.VerifyMismatchError:
return render_template("login.html", error="Invalid username or password"), 401
if page is None:
page = "/"
token = createUserSession(username)
resp = make_response(redirect(page))
resp.set_cookie("lnsession", token, max_age=app.config["SESSION_TIMEOUT"])
return resp
@app.route("/login", methods=["GET"])
def login_get():
return render_template("login.html")
@app.route("/logout")
def logout():
session_id = request.cookies.get("lnsession")
if verifiedSessions.get(session_id):
expireSession(session_id)
resp = make_response(redirect("/"))
resp.set_cookie("lnsession", "", expires=0)
return resp
@app.route("/api_logged_in")
def api_logged_in():
session_id = request.cookies.get("lnsession")
if verifiedSessions.get(session_id):
return {"logged_in": True}
return {"logged_in": False}
|
