331 lines
12 KiB
Python
331 lines
12 KiB
Python
import os
|
|
import re
|
|
import sqlite3
|
|
from base64 import b64encode
|
|
from io import BytesIO
|
|
from datetime import datetime, timezone
|
|
from functools import wraps
|
|
from pathlib import Path
|
|
|
|
from flask import Flask, abort, flash, g, redirect, render_template, request, session, url_for
|
|
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"))
|
|
|
|
app = Flask(__name__)
|
|
app.config.update(
|
|
SECRET_KEY=os.environ.get("SECRET_KEY", "change-this-secret-before-production"),
|
|
DATABASE=DATABASE,
|
|
)
|
|
|
|
|
|
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,
|
|
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,
|
|
author_id INTEGER NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL,
|
|
FOREIGN KEY (author_id) REFERENCES users(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",
|
|
}
|
|
for column, statement in migrations.items():
|
|
if column not in user_columns:
|
|
db.execute(statement)
|
|
|
|
# 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
|
|
|
|
|
|
@app.context_processor
|
|
def inject_current_user():
|
|
user = None
|
|
if "user_id" in session:
|
|
user = get_db().execute("SELECT id, username, 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 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 FROM posts JOIN users ON users.id = posts.author_id
|
|
WHERE posts.slug = ?""",
|
|
(slug,),
|
|
).fetchone()
|
|
if article is None:
|
|
abort(404)
|
|
return render_template("post.html", post=article)
|
|
|
|
|
|
@app.route("/register", methods=("GET", "POST"))
|
|
def register():
|
|
if request.method == "POST":
|
|
username = request.form.get("username", "").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")
|
|
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()),
|
|
)
|
|
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")
|
|
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")
|
|
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.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, 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.route("/write", methods=("GET", "POST"))
|
|
@login_required
|
|
def write():
|
|
if request.method == "POST":
|
|
try:
|
|
title, slug, excerpt, body, category, now = post_from_request()
|
|
get_db().execute(
|
|
"""INSERT INTO posts (title, slug, excerpt, body, category, author_id, created_at, updated_at)
|
|
VALUES (?, ?, ?, ?, ?, ?, ?, ?)""",
|
|
(title, slug, excerpt, body, category, session["user_id"], now, now),
|
|
)
|
|
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"])
|
|
get_db().execute(
|
|
"""UPDATE posts SET title=?, slug=?, excerpt=?, body=?, category=?, updated_at=? WHERE id=?""",
|
|
(title, new_slug, excerpt, body, category, now, article["id"]),
|
|
)
|
|
get_db().commit()
|
|
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)
|