thuttu/docker-ep-web-server
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add partial email support and todo list

Browse Source
This commit is contained in:
thuttu
2026-08-29 19:57:31 +03:00
parent d93690a36b
commit 8a2cbe5b8b
12 changed files with 187 additions and 10 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+138 -5
View File
@@ -1,11 +1,15 @@
import os
import re
import sqlite3
import smtplib
from base64 import b64encode
from io import BytesIO
from datetime import datetime, timezone
from datetime import datetime, timedelta, timezone
from email.message import EmailMessage
from functools import wraps
from pathlib import Path
from secrets import token_urlsafe
from hashlib import sha256
from flask import Flask, abort, flash, g, redirect, render_template, request, session, url_for
import pyotp
@@ -19,6 +23,12 @@ app = Flask(__name__)
app.config.update(
SECRET_KEY=os.environ.get("SECRET_KEY", "change-this-secret-before-production"),
DATABASE=DATABASE,
MAIL_HOST=os.environ.get("MAIL_HOST"),
MAIL_PORT=int(os.environ.get("MAIL_PORT", "587")),
MAIL_USERNAME=os.environ.get("MAIL_USERNAME"),
MAIL_PASSWORD=os.environ.get("MAIL_PASSWORD"),
MAIL_FROM=os.environ.get("MAIL_FROM", "noreply@eternityproject.fi"),
PUBLIC_URL=os.environ.get("PUBLIC_URL", "http://localhost:8000").rstrip("/"),
)
@@ -44,6 +54,7 @@ def init_db():
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT UNIQUE NOT NULL,
email TEXT UNIQUE,
password_hash TEXT NOT NULL,
created_at TEXT NOT NULL,
is_approved INTEGER NOT NULL DEFAULT 0,
@@ -63,6 +74,15 @@ def init_db():
updated_at TEXT NOT NULL,
FOREIGN KEY (author_id) REFERENCES users(id)
);
CREATE TABLE IF NOT EXISTS password_reset_tokens (
id INTEGER PRIMARY KEY AUTOINCREMENT,
user_id INTEGER NOT NULL,
token_hash TEXT UNIQUE NOT NULL,
expires_at TEXT NOT NULL,
used_at TEXT,
created_at TEXT NOT NULL,
FOREIGN KEY (user_id) REFERENCES users(id)
);
"""
)
user_columns = {row["name"] for row in db.execute("PRAGMA table_info(users)")}
@@ -71,10 +91,12 @@ def init_db():
"role": "ALTER TABLE users ADD COLUMN role TEXT NOT NULL DEFAULT 'member'",
"mfa_secret": "ALTER TABLE users ADD COLUMN mfa_secret TEXT",
"mfa_enabled": "ALTER TABLE users ADD COLUMN mfa_enabled INTEGER NOT NULL DEFAULT 0",
"email": "ALTER TABLE users ADD COLUMN email TEXT",
}
for column, statement in migrations.items():
if column not in user_columns:
db.execute(statement)
db.execute("CREATE UNIQUE INDEX IF NOT EXISTS users_email_unique ON users(email)")
# Preserve access for accounts created before approvals were introduced.
db.execute("UPDATE users SET is_approved = 1 WHERE role = 'member' AND is_approved = 0")
@@ -135,6 +157,40 @@ def post_from_request(post_id=None):
return title, unique_slug(title, post_id), excerpt, body, category, now
def valid_email(email):
return re.fullmatch(r"[^\s@]+@[^\s@]+\.[^\s@]+", email) is not None
def create_reset_token(user_id):
token = token_urlsafe(32)
now = datetime.now(timezone.utc)
get_db().execute("UPDATE password_reset_tokens SET used_at = ? WHERE user_id = ? AND used_at IS NULL", (now.isoformat(), user_id))
get_db().execute(
"""INSERT INTO password_reset_tokens (user_id, token_hash, expires_at, created_at)
VALUES (?, ?, ?, ?)""",
(user_id, sha256(token.encode("utf-8")).hexdigest(), (now + timedelta(hours=1)).isoformat(), now.isoformat()),
)
get_db().commit()
return token
def send_reset_email(user):
if not app.config["MAIL_HOST"]:
raise RuntimeError("Email is not configured. Set MAIL_HOST and related SMTP settings.")
token = create_reset_token(user["id"])
reset_url = f"{app.config['PUBLIC_URL']}{url_for('reset_password', token=token)}"
message = EmailMessage()
message["Subject"] = "Reset your Eternity Project password"
message["From"] = app.config["MAIL_FROM"]
message["To"] = user["email"]
message.set_content(f"Hello {user['username']},\n\nUse this link within one hour to choose a new password:\n{reset_url}\n\nIf you did not request this, you can ignore this email.")
with smtplib.SMTP(app.config["MAIL_HOST"], app.config["MAIL_PORT"]) as client:
client.starttls()
if app.config["MAIL_USERNAME"]:
client.login(app.config["MAIL_USERNAME"], app.config["MAIL_PASSWORD"])
client.send_message(message)
@app.context_processor
def inject_current_user():
user = None
@@ -168,22 +224,25 @@ def post(slug):
def register():
if request.method == "POST":
username = request.form.get("username", "").strip().lower()
email = request.form.get("email", "").strip().lower()
password = request.form.get("password", "")
if not re.fullmatch(r"[a-z0-9_-]{3,32}", username):
flash("Use 3-32 lowercase letters, numbers, hyphens, or underscores.", "error")
elif len(password) < 10:
flash("Choose a password with at least 10 characters.", "error")
elif not valid_email(email):
flash("Enter a valid email address.", "error")
else:
try:
get_db().execute(
"INSERT INTO users (username, password_hash, created_at, is_approved) VALUES (?, ?, ?, 0)",
(username, generate_password_hash(password), datetime.now(timezone.utc).isoformat()),
"INSERT INTO users (username, email, password_hash, created_at, is_approved) VALUES (?, ?, ?, ?, 0)",
(username, email, generate_password_hash(password), datetime.now(timezone.utc).isoformat()),
)
get_db().commit()
flash("Registration received. An administrator must approve it before you can sign in.", "success")
return redirect(url_for("login"))
except sqlite3.IntegrityError:
flash("That handle is already in use.", "error")
flash("That handle or email address is already in use.", "error")
return render_template("auth.html", mode="register")
@@ -206,6 +265,64 @@ def login():
return render_template("auth.html", mode="login")
@app.route("/password/forgot", methods=("GET", "POST"))
def forgot_password():
if request.method == "POST":
email = request.form.get("email", "").strip().lower()
user = get_db().execute("SELECT * FROM users WHERE email = ?", (email,)).fetchone()
if user is not None:
try:
send_reset_email(user)
except (RuntimeError, OSError, smtplib.SMTPException):
app.logger.exception("Unable to send password reset email")
flash("If that address belongs to an account, a reset link has been sent.", "success")
return redirect(url_for("login"))
return render_template("password_form.html", mode="forgot")
@app.route("/password/reset/<token>", methods=("GET", "POST"))
def reset_password(token):
now = datetime.now(timezone.utc).isoformat()
reset = get_db().execute(
"""SELECT * FROM password_reset_tokens WHERE token_hash = ? AND used_at IS NULL AND expires_at > ?""",
(sha256(token.encode("utf-8")).hexdigest(), now),
).fetchone()
if reset is None:
flash("That password reset link is invalid or has expired.", "error")
return redirect(url_for("forgot_password"))
if request.method == "POST":
password = request.form.get("password", "")
if len(password) < 10:
flash("Choose a password with at least 10 characters.", "error")
else:
db = get_db()
db.execute("UPDATE users SET password_hash = ? WHERE id = ?", (generate_password_hash(password), reset["user_id"]))
db.execute("UPDATE password_reset_tokens SET used_at = ? WHERE id = ?", (now, reset["id"]))
db.commit()
flash("Password reset. Sign in with your new password.", "success")
return redirect(url_for("login"))
return render_template("password_form.html", mode="reset")
@app.route("/account/password", methods=("GET", "POST"))
@login_required
def change_password():
if request.method == "POST":
user = get_db().execute("SELECT password_hash FROM users WHERE id = ?", (session["user_id"],)).fetchone()
current_password = request.form.get("current_password", "")
new_password = request.form.get("password", "")
if not check_password_hash(user["password_hash"], current_password):
flash("Your current password is incorrect.", "error")
elif len(new_password) < 10:
flash("Choose a new password with at least 10 characters.", "error")
else:
get_db().execute("UPDATE users SET password_hash = ? WHERE id = ?", (generate_password_hash(new_password), session["user_id"]))
get_db().commit()
flash("Password updated.", "success")
return redirect(url_for("index"))
return render_template("password_form.html", mode="change")
@app.post("/logout")
def logout():
session.clear()
@@ -268,7 +385,7 @@ def mfa_verify():
@admin_required
def admin_users():
users = get_db().execute(
"SELECT id, username, created_at, is_approved, mfa_enabled, role FROM users ORDER BY is_approved, created_at DESC"
"SELECT id, username, email, created_at, is_approved, mfa_enabled, role FROM users ORDER BY is_approved, created_at DESC"
).fetchall()
return render_template("admin_users.html", users=users)
@@ -282,6 +399,22 @@ def approve_user(user_id):
return redirect(url_for("admin_users"))
@app.post("/admin/users/<int:user_id>/password-reset")
@admin_required
def admin_password_reset(user_id):
user = get_db().execute("SELECT * FROM users WHERE id = ? AND role = 'member'", (user_id,)).fetchone()
if user is None or not user["email"]:
flash("That member does not have an email address for password recovery.", "error")
else:
try:
send_reset_email(user)
flash(f"Password reset email sent to {user['email']}.", "success")
except (RuntimeError, OSError, smtplib.SMTPException):
app.logger.exception("Unable to send administrator password reset email")
flash("Password reset email could not be sent. Check SMTP settings.", "error")
return redirect(url_for("admin_users"))
@app.route("/write", methods=("GET", "POST"))
@login_required
def write():