720 lines
31 KiB
Python
720 lines
31 KiB
Python
import os
|
|
import re
|
|
import sqlite3
|
|
import smtplib
|
|
from base64 import b64encode
|
|
from io import BytesIO
|
|
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 uuid import uuid4
|
|
|
|
from flask import Flask, abort, flash, g, redirect, render_template, request, send_from_directory, session, url_for
|
|
from flask_wtf.csrf import CSRFProtect
|
|
from PIL import Image, ImageOps, UnidentifiedImageError
|
|
import pyotp
|
|
import qrcode
|
|
from werkzeug.security import check_password_hash, generate_password_hash
|
|
|
|
BASE_DIR = Path(__file__).resolve().parent
|
|
DATABASE = Path(os.environ.get("DATABASE_PATH", BASE_DIR / "data" / "eternity.db"))
|
|
UPLOAD_FOLDER = Path(os.environ.get("UPLOAD_FOLDER", DATABASE.parent / "uploads"))
|
|
|
|
app = Flask(__name__)
|
|
app.config.update(
|
|
SECRET_KEY=os.environ.get("SECRET_KEY", "change-this-secret-before-production"),
|
|
DATABASE=DATABASE,
|
|
UPLOAD_FOLDER=UPLOAD_FOLDER,
|
|
MAX_CONTENT_LENGTH=8 * 1024 * 1024,
|
|
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("/"),
|
|
)
|
|
csrf = CSRFProtect(app)
|
|
|
|
|
|
def get_db():
|
|
if "db" not in g:
|
|
app.config["DATABASE"].parent.mkdir(parents=True, exist_ok=True)
|
|
g.db = sqlite3.connect(app.config["DATABASE"])
|
|
g.db.row_factory = sqlite3.Row
|
|
return g.db
|
|
|
|
|
|
@app.teardown_appcontext
|
|
def close_db(_error=None):
|
|
db = g.pop("db", None)
|
|
if db is not None:
|
|
db.close()
|
|
|
|
|
|
def init_db():
|
|
db = get_db()
|
|
db.executescript(
|
|
"""
|
|
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,
|
|
role TEXT NOT NULL DEFAULT 'member',
|
|
mfa_secret TEXT,
|
|
mfa_enabled INTEGER NOT NULL DEFAULT 0
|
|
);
|
|
CREATE TABLE IF NOT EXISTS posts (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
title TEXT NOT NULL,
|
|
slug TEXT UNIQUE NOT NULL,
|
|
excerpt TEXT NOT NULL,
|
|
body TEXT NOT NULL,
|
|
category TEXT NOT NULL,
|
|
image_filename TEXT,
|
|
author_id INTEGER NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
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)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS email_verification_tokens (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
user_id INTEGER NOT NULL,
|
|
email_address TEXT NOT NULL,
|
|
token_hash TEXT UNIQUE NOT NULL,
|
|
token_type TEXT NOT NULL,
|
|
expires_at TEXT NOT NULL,
|
|
used_at TEXT,
|
|
created_at TEXT NOT NULL,
|
|
FOREIGN KEY (user_id) REFERENCES users(id)
|
|
);
|
|
CREATE TABLE IF NOT EXISTS post_images (
|
|
id INTEGER PRIMARY KEY AUTOINCREMENT,
|
|
post_id INTEGER NOT NULL,
|
|
filename TEXT UNIQUE NOT NULL,
|
|
caption TEXT NOT NULL DEFAULT '',
|
|
position INTEGER NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
FOREIGN KEY (post_id) REFERENCES posts(id)
|
|
);
|
|
"""
|
|
)
|
|
user_columns = {row["name"] for row in db.execute("PRAGMA table_info(users)")}
|
|
migrations = {
|
|
"is_approved": "ALTER TABLE users ADD COLUMN is_approved INTEGER NOT NULL DEFAULT 0",
|
|
"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",
|
|
"email_verified": "ALTER TABLE users ADD COLUMN email_verified INTEGER NOT NULL DEFAULT 0",
|
|
"real_name": "ALTER TABLE users ADD COLUMN real_name TEXT",
|
|
}
|
|
post_columns = {row["name"] for row in db.execute("PRAGMA table_info(posts)")}
|
|
for column, statement in migrations.items():
|
|
if column not in user_columns:
|
|
db.execute(statement)
|
|
if column == "email_verified":
|
|
db.execute("UPDATE users SET email_verified = 1 WHERE email IS NOT NULL")
|
|
if "image_filename" not in post_columns:
|
|
db.execute("ALTER TABLE posts ADD COLUMN image_filename TEXT")
|
|
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")
|
|
admin = db.execute("SELECT id FROM users WHERE username = 'admin'").fetchone()
|
|
if admin is None:
|
|
db.execute(
|
|
"""INSERT INTO users (username, password_hash, created_at, is_approved, role)
|
|
VALUES (?, ?, ?, 1, 'admin')""",
|
|
("admin", generate_password_hash("admin"), datetime.now(timezone.utc).isoformat()),
|
|
)
|
|
db.commit()
|
|
|
|
|
|
def login_required(view):
|
|
@wraps(view)
|
|
def wrapped_view(*args, **kwargs):
|
|
if "user_id" not in session or not session.get("mfa_verified"):
|
|
flash("Sign in to continue.", "notice")
|
|
return redirect(url_for("login", next=request.path))
|
|
return view(*args, **kwargs)
|
|
|
|
return wrapped_view
|
|
|
|
|
|
def admin_required(view):
|
|
@wraps(view)
|
|
@login_required
|
|
def wrapped_view(*args, **kwargs):
|
|
user = get_db().execute("SELECT role FROM users WHERE id = ?", (session["user_id"],)).fetchone()
|
|
if user is None or user["role"] != "admin":
|
|
abort(403)
|
|
return view(*args, **kwargs)
|
|
|
|
return wrapped_view
|
|
|
|
|
|
def unique_slug(title, post_id=None):
|
|
base = re.sub(r"[^a-z0-9]+", "-", title.lower()).strip("-") or "untitled"
|
|
slug = base
|
|
counter = 2
|
|
db = get_db()
|
|
while True:
|
|
row = db.execute("SELECT id FROM posts WHERE slug = ?", (slug,)).fetchone()
|
|
if row is None or row["id"] == post_id:
|
|
return slug
|
|
slug = f"{base}-{counter}"
|
|
counter += 1
|
|
|
|
|
|
def post_from_request(post_id=None):
|
|
title = request.form.get("title", "").strip()
|
|
excerpt = request.form.get("excerpt", "").strip()
|
|
body = request.form.get("body", "").strip()
|
|
category = request.form.get("category", "Engineering").strip() or "Engineering"
|
|
if not title or not excerpt or not body:
|
|
raise ValueError("Title, summary, and article body are required.")
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
return title, unique_slug(title, post_id), excerpt, body, category, now
|
|
|
|
|
|
def save_post_image(image_file):
|
|
if image_file is None or not image_file.filename:
|
|
return None
|
|
try:
|
|
image = Image.open(image_file.stream)
|
|
image.verify()
|
|
source_format = image.format
|
|
image_file.stream.seek(0)
|
|
image = ImageOps.exif_transpose(Image.open(image_file.stream)).convert("RGB")
|
|
except (UnidentifiedImageError, OSError, SyntaxError) as error:
|
|
raise ValueError("Upload a valid JPEG, PNG, or WebP image.") from error
|
|
if source_format not in {"JPEG", "PNG", "WEBP"}:
|
|
raise ValueError("Upload a JPEG, PNG, or WebP image.")
|
|
image.thumbnail((2400, 2400))
|
|
filename = f"{uuid4().hex}.webp"
|
|
app.config["UPLOAD_FOLDER"].mkdir(parents=True, exist_ok=True)
|
|
image.save(app.config["UPLOAD_FOLDER"] / filename, "WEBP", quality=88, method=6)
|
|
return filename
|
|
|
|
|
|
def save_inline_images(post_id):
|
|
image_files = [image_file for image_file in request.files.getlist("images") if image_file.filename]
|
|
captions = request.form.getlist("image_captions")
|
|
if not image_files:
|
|
return
|
|
next_position = get_db().execute(
|
|
"SELECT COALESCE(MAX(position), 0) + 1 FROM post_images WHERE post_id = ?", (post_id,)
|
|
).fetchone()[0]
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
for offset, image_file in enumerate(image_files):
|
|
filename = save_post_image(image_file)
|
|
caption = captions[offset].strip() if offset < len(captions) else ""
|
|
get_db().execute(
|
|
"INSERT INTO post_images (post_id, filename, caption, position, created_at) VALUES (?, ?, ?, ?, ?)",
|
|
(post_id, filename, caption, next_position + offset, now),
|
|
)
|
|
|
|
|
|
def article_blocks(body, images):
|
|
image_map = {str(index): image for index, image in enumerate(images, start=1)}
|
|
blocks = []
|
|
marker = re.compile(
|
|
r"```(?P<language>[A-Za-z0-9][A-Za-z0-9_-]*)?[ \t]*\r?\n(?P<code>.*?)(?:\r?\n)?```|\[\[image:(?P<image>\d+)\]\]",
|
|
re.DOTALL,
|
|
)
|
|
cursor = 0
|
|
for match in marker.finditer(body):
|
|
if match.start() > cursor:
|
|
blocks.append(("text", body[cursor:match.start()]))
|
|
if match.group("code") is not None:
|
|
blocks.append(("code", {"language": match.group("language") or "text", "content": match.group("code")}))
|
|
else:
|
|
image = image_map.get(match.group("image"))
|
|
if image is not None:
|
|
blocks.append(("image", image))
|
|
else:
|
|
blocks.append(("text", match.group(0)))
|
|
cursor = match.end()
|
|
if cursor < len(body):
|
|
blocks.append(("text", body[cursor:]))
|
|
return blocks
|
|
|
|
|
|
def valid_email(email):
|
|
return re.fullmatch(r"[^\s@]+@[^\s@]+\.[^\s@]+", email) is not None
|
|
|
|
|
|
def normalize_real_name(real_name):
|
|
return " ".join(real_name.split())
|
|
|
|
|
|
def valid_real_name(real_name):
|
|
return 2 <= len(real_name) <= 100 and all(character.isalnum() or character in " .'-" for character in real_name)
|
|
|
|
|
|
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)
|
|
|
|
|
|
def create_email_verification_token(user_id, email_address, token_type):
|
|
token = token_urlsafe(32)
|
|
now = datetime.now(timezone.utc)
|
|
db = get_db()
|
|
db.execute(
|
|
"UPDATE email_verification_tokens SET used_at = ? WHERE user_id = ? AND token_type = ? AND used_at IS NULL",
|
|
(now.isoformat(), user_id, token_type),
|
|
)
|
|
db.execute(
|
|
"""INSERT INTO email_verification_tokens (user_id, email_address, token_hash, token_type, expires_at, created_at)
|
|
VALUES (?, ?, ?, ?, ?, ?)""",
|
|
(user_id, email_address, sha256(token.encode("utf-8")).hexdigest(), token_type, (now + timedelta(hours=24)).isoformat(), now.isoformat()),
|
|
)
|
|
db.commit()
|
|
return token
|
|
|
|
|
|
def send_verification_email(user, email_address, token_type):
|
|
if not app.config["MAIL_HOST"]:
|
|
raise RuntimeError("Email is not configured. Set MAIL_HOST and related SMTP settings.")
|
|
token = create_email_verification_token(user["id"], email_address, token_type)
|
|
verification_url = f"{app.config['PUBLIC_URL']}{url_for('verify_email', token=token)}"
|
|
action = "confirm this email address for your account" if token_type == "registration" else "confirm this new email address"
|
|
message = EmailMessage()
|
|
message["Subject"] = "Confirm your Eternity Project email address"
|
|
message["From"] = app.config["MAIL_FROM"]
|
|
message["To"] = email_address
|
|
message.set_content(f"Hello {user['username']},\n\nUse this link within 24 hours to {action}:\n{verification_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
|
|
if "user_id" in session:
|
|
user = get_db().execute("SELECT id, username, real_name, role FROM users WHERE id = ?", (session["user_id"],)).fetchone()
|
|
return {"current_user": user}
|
|
|
|
|
|
@app.route("/")
|
|
def index():
|
|
posts = get_db().execute(
|
|
"""SELECT posts.*, users.username, users.real_name FROM posts JOIN users ON users.id = posts.author_id
|
|
ORDER BY posts.created_at DESC"""
|
|
).fetchall()
|
|
return render_template("index.html", posts=posts)
|
|
|
|
|
|
@app.route("/post/<slug>")
|
|
def post(slug):
|
|
article = get_db().execute(
|
|
"""SELECT posts.*, users.username, users.real_name FROM posts JOIN users ON users.id = posts.author_id
|
|
WHERE posts.slug = ?""",
|
|
(slug,),
|
|
).fetchone()
|
|
if article is None:
|
|
abort(404)
|
|
images = get_db().execute("SELECT * FROM post_images WHERE post_id = ? ORDER BY position", (article["id"],)).fetchall()
|
|
return render_template("post.html", post=article, blocks=article_blocks(article["body"], images))
|
|
|
|
|
|
@app.get("/uploads/<path:filename>")
|
|
def uploaded_image(filename):
|
|
return send_from_directory(app.config["UPLOAD_FOLDER"], filename)
|
|
|
|
|
|
@app.route("/register", methods=("GET", "POST"))
|
|
def register():
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").strip().lower()
|
|
real_name = normalize_real_name(request.form.get("real_name", ""))
|
|
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 not valid_real_name(real_name):
|
|
flash("Enter a real name using 2-100 letters, numbers, spaces, hyphens, apostrophes, or periods.", "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:
|
|
cursor = get_db().execute(
|
|
"INSERT INTO users (username, real_name, email, password_hash, created_at, is_approved) VALUES (?, ?, ?, ?, ?, 0)",
|
|
(username, real_name, email, generate_password_hash(password), datetime.now(timezone.utc).isoformat()),
|
|
)
|
|
get_db().commit()
|
|
user = {"id": cursor.lastrowid, "username": username}
|
|
try:
|
|
send_verification_email(user, email, "registration")
|
|
flash("Check your email to verify your address. An administrator must also approve your account before you can sign in.", "success")
|
|
except (RuntimeError, OSError, smtplib.SMTPException):
|
|
app.logger.exception("Unable to send registration verification email")
|
|
flash("Registration received, but the verification email could not be sent. Contact an administrator.", "error")
|
|
return redirect(url_for("login"))
|
|
except sqlite3.IntegrityError:
|
|
flash("That handle or email address is already in use.", "error")
|
|
return render_template("auth.html", mode="register")
|
|
|
|
|
|
@app.route("/login", methods=("GET", "POST"))
|
|
def login():
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").strip().lower()
|
|
user = get_db().execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
|
|
if user is None or not check_password_hash(user["password_hash"], request.form.get("password", "")):
|
|
flash("Invalid handle or password.", "error")
|
|
elif not user["is_approved"]:
|
|
flash("Your account is awaiting administrator approval.", "notice")
|
|
elif user["email"] and not user["email_verified"]:
|
|
flash("Verify your email address before you can sign in.", "notice")
|
|
else:
|
|
session.clear()
|
|
session["mfa_pending_user_id"] = user["id"]
|
|
session["login_next"] = request.args.get("next") or url_for("index")
|
|
if user["mfa_enabled"]:
|
|
return redirect(url_for("mfa_verify"))
|
|
return redirect(url_for("mfa_setup"))
|
|
return render_template("auth.html", mode="login")
|
|
|
|
|
|
@app.route("/email/resend", methods=("GET", "POST"))
|
|
def resend_verification_email():
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").strip().lower()
|
|
user = get_db().execute("SELECT * FROM users WHERE username = ?", (username,)).fetchone()
|
|
if user is not None and user["email"] and not user["email_verified"] and check_password_hash(user["password_hash"], request.form.get("password", "")):
|
|
try:
|
|
send_verification_email(user, user["email"], "registration")
|
|
except (RuntimeError, OSError, smtplib.SMTPException):
|
|
app.logger.exception("Unable to resend verification email")
|
|
flash("If those credentials belong to an unverified account, a verification link has been sent.", "success")
|
|
return redirect(url_for("login"))
|
|
return render_template("auth.html", mode="resend")
|
|
|
|
|
|
@app.route("/email/verify/<token>", methods=("GET", "POST"))
|
|
def verify_email(token):
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
verification = get_db().execute(
|
|
"""SELECT * FROM email_verification_tokens
|
|
WHERE token_hash = ? AND used_at IS NULL AND expires_at > ?""",
|
|
(sha256(token.encode("utf-8")).hexdigest(), now),
|
|
).fetchone()
|
|
if verification is None:
|
|
flash("That email verification link is invalid or has expired.", "error")
|
|
return redirect(url_for("login"))
|
|
if request.method == "POST":
|
|
db = get_db()
|
|
try:
|
|
if verification["token_type"] == "registration":
|
|
db.execute("UPDATE users SET email_verified = 1 WHERE id = ?", (verification["user_id"],))
|
|
message = "Email verified. You can sign in once an administrator approves your account."
|
|
else:
|
|
db.execute("UPDATE users SET email = ?, email_verified = 1 WHERE id = ?", (verification["email_address"], verification["user_id"]))
|
|
message = "Email address updated and verified."
|
|
db.execute("UPDATE email_verification_tokens SET used_at = ? WHERE id = ?", (now, verification["id"]))
|
|
db.commit()
|
|
except sqlite3.IntegrityError:
|
|
db.rollback()
|
|
flash("That email address is already in use. Request a verification link for a different address.", "error")
|
|
return redirect(url_for("change_email"))
|
|
flash(message, "success")
|
|
return redirect(url_for("login"))
|
|
return render_template("email_verify.html", token=token, email=verification["email_address"], token_type=verification["token_type"])
|
|
|
|
|
|
@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.route("/account/profile", methods=("GET", "POST"))
|
|
@login_required
|
|
def change_profile():
|
|
user = get_db().execute("SELECT real_name FROM users WHERE id = ?", (session["user_id"],)).fetchone()
|
|
if request.method == "POST":
|
|
real_name = normalize_real_name(request.form.get("real_name", ""))
|
|
if not valid_real_name(real_name):
|
|
flash("Enter a real name using 2-100 letters, numbers, spaces, hyphens, apostrophes, or periods.", "error")
|
|
else:
|
|
get_db().execute("UPDATE users SET real_name = ? WHERE id = ?", (real_name, session["user_id"]))
|
|
get_db().commit()
|
|
flash("Profile updated.", "success")
|
|
return redirect(url_for("index"))
|
|
return render_template("profile_form.html", real_name=user["real_name"] or "")
|
|
|
|
|
|
@app.route("/account/email", methods=("GET", "POST"))
|
|
@login_required
|
|
def change_email():
|
|
user = get_db().execute("SELECT id, username, email, password_hash FROM users WHERE id = ?", (session["user_id"],)).fetchone()
|
|
if request.method == "POST":
|
|
email = request.form.get("email", "").strip().lower()
|
|
current_password = request.form.get("current_password", "")
|
|
if not check_password_hash(user["password_hash"], current_password):
|
|
flash("Your current password is incorrect.", "error")
|
|
elif not valid_email(email):
|
|
flash("Enter a valid email address.", "error")
|
|
elif email == user["email"]:
|
|
flash("That is already your email address.", "notice")
|
|
else:
|
|
try:
|
|
existing_user = get_db().execute("SELECT id FROM users WHERE email = ? AND id != ?", (email, session["user_id"])).fetchone()
|
|
if existing_user is not None:
|
|
flash("That email address is already in use.", "error")
|
|
return render_template("email_form.html", email=user["email"])
|
|
send_verification_email(user, email, "email_change")
|
|
flash("Verification email sent. Confirm the new address within 24 hours to complete the change.", "success")
|
|
return redirect(url_for("index"))
|
|
except (RuntimeError, OSError, smtplib.SMTPException):
|
|
app.logger.exception("Unable to send email change verification email")
|
|
flash("Verification email could not be sent. Check SMTP settings and try again.", "error")
|
|
return render_template("email_form.html", email=user["email"])
|
|
|
|
|
|
@app.post("/logout")
|
|
def logout():
|
|
session.clear()
|
|
return redirect(url_for("index"))
|
|
|
|
|
|
def pending_mfa_user():
|
|
user_id = session.get("mfa_pending_user_id")
|
|
if user_id is None:
|
|
return None
|
|
return get_db().execute("SELECT * FROM users WHERE id = ?", (user_id,)).fetchone()
|
|
|
|
|
|
def mfa_qr_code(secret, username):
|
|
uri = pyotp.TOTP(secret).provisioning_uri(name=username, issuer_name="Eternity Project")
|
|
image = qrcode.make(uri)
|
|
buffer = BytesIO()
|
|
image.save(buffer, format="PNG")
|
|
return f"data:image/png;base64,{b64encode(buffer.getvalue()).decode('ascii')}"
|
|
|
|
|
|
@app.route("/mfa/setup", methods=("GET", "POST"))
|
|
def mfa_setup():
|
|
user = pending_mfa_user()
|
|
if user is None or user["mfa_enabled"]:
|
|
return redirect(url_for("login"))
|
|
secret = session.setdefault("mfa_setup_secret", pyotp.random_base32())
|
|
if request.method == "POST":
|
|
if pyotp.TOTP(secret).verify(request.form.get("code", ""), valid_window=1):
|
|
get_db().execute("UPDATE users SET mfa_secret = ?, mfa_enabled = 1 WHERE id = ?", (secret, user["id"]))
|
|
get_db().commit()
|
|
next_url = session.get("login_next", url_for("index"))
|
|
session.clear()
|
|
session["user_id"] = user["id"]
|
|
session["mfa_verified"] = True
|
|
return redirect(next_url)
|
|
flash("That verification code was not accepted. Try the current code.", "error")
|
|
return render_template(
|
|
"mfa.html", mode="setup", secret=secret, username=user["username"], qr_code=mfa_qr_code(secret, user["username"])
|
|
)
|
|
|
|
|
|
@app.route("/mfa/verify", methods=("GET", "POST"))
|
|
def mfa_verify():
|
|
user = pending_mfa_user()
|
|
if user is None or not user["mfa_enabled"] or not user["mfa_secret"]:
|
|
return redirect(url_for("login"))
|
|
if request.method == "POST":
|
|
if pyotp.TOTP(user["mfa_secret"]).verify(request.form.get("code", ""), valid_window=1):
|
|
next_url = session.get("login_next", url_for("index"))
|
|
session.clear()
|
|
session["user_id"] = user["id"]
|
|
session["mfa_verified"] = True
|
|
return redirect(next_url)
|
|
flash("That verification code was not accepted. Try the current code.", "error")
|
|
return render_template("mfa.html", mode="verify", username=user["username"])
|
|
|
|
|
|
@app.get("/admin/users")
|
|
@admin_required
|
|
def admin_users():
|
|
users = get_db().execute(
|
|
"SELECT id, username, real_name, email, email_verified, 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)
|
|
|
|
|
|
@app.post("/admin/users/<int:user_id>/approve")
|
|
@admin_required
|
|
def approve_user(user_id):
|
|
cursor = get_db().execute("UPDATE users SET is_approved = 1 WHERE id = ? AND role = 'member'", (user_id,))
|
|
get_db().commit()
|
|
flash("Account approved." if cursor.rowcount else "No pending member account was changed.", "success")
|
|
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():
|
|
if request.method == "POST":
|
|
try:
|
|
title, slug, excerpt, body, category, now = post_from_request()
|
|
image_file = request.files.get("image")
|
|
image_filename = save_post_image(image_file) if image_file and image_file.filename else None
|
|
cursor = get_db().execute(
|
|
"""INSERT INTO posts (title, slug, excerpt, body, category, image_filename, author_id, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(title, slug, excerpt, body, category, image_filename, session["user_id"], now, now),
|
|
)
|
|
save_inline_images(cursor.lastrowid)
|
|
get_db().commit()
|
|
return redirect(url_for("post", slug=slug))
|
|
except ValueError as error:
|
|
flash(str(error), "error")
|
|
return render_template("editor.html", post=None)
|
|
|
|
|
|
@app.route("/post/<slug>/edit", methods=("GET", "POST"))
|
|
@login_required
|
|
def edit_post(slug):
|
|
article = get_db().execute("SELECT * FROM posts WHERE slug = ?", (slug,)).fetchone()
|
|
if article is None:
|
|
abort(404)
|
|
if article["author_id"] != session["user_id"]:
|
|
abort(403)
|
|
if request.method == "POST":
|
|
try:
|
|
title, new_slug, excerpt, body, category, now = post_from_request(article["id"])
|
|
image_file = request.files.get("image")
|
|
image_filename = save_post_image(image_file) if image_file and image_file.filename else article["image_filename"]
|
|
get_db().execute(
|
|
"""UPDATE posts SET title=?, slug=?, excerpt=?, body=?, category=?, image_filename=?, updated_at=? WHERE id=?""",
|
|
(title, new_slug, excerpt, body, category, image_filename, now, article["id"]),
|
|
)
|
|
save_inline_images(article["id"])
|
|
get_db().commit()
|
|
if image_filename != article["image_filename"] and article["image_filename"]:
|
|
(app.config["UPLOAD_FOLDER"] / article["image_filename"]).unlink(missing_ok=True)
|
|
return redirect(url_for("post", slug=new_slug))
|
|
except ValueError as error:
|
|
flash(str(error), "error")
|
|
return render_template("editor.html", post=article)
|
|
|
|
|
|
with app.app_context():
|
|
init_db()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
app.run(host="0.0.0.0", port=8000)
|