first commit

2026-08-29 10:35:34 +03:00
commit 1089c784e9
17 changed files with 543 additions and 0 deletions
@@ -0,0 +1,317 @@
+import os
+import re
+import sqlite3
+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
+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()
+
+
+@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"])
+
+
+@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)