PRIVATE_SELF_DEV/Approval-Workflow
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
main
Approval-Workflow/app.py
T

580 lines
25 KiB
Python
Raw Permalink Normal View History

v_9.3
2026-09-13 20:03:24 +01:00
"""ApprovalFlow — self-hosted email approval workflows."""
import functools
import json
import os
import re
import secrets
from flask import (Flask, abort, flash, g, redirect, render_template,
request, send_file, session, url_for)
from werkzeug.security import check_password_hash, generate_password_hash
from werkzeug.utils import secure_filename
import db
import pipeline
import totp
from mailer import send_mail
from db import DEFAULT_TEMPLATES
app = Flask(__name__)
app.secret_key = os.environ.get("SECRET_KEY") or secrets.token_hex(32)
# 15 MB attachment limit (+ small allowance for the rest of the form)
app.config["MAX_CONTENT_LENGTH"] = db.MAX_ATTACHMENT_BYTES + 512 * 1024
@app.errorhandler(413)
def too_large(e):
flash("The uploaded files are too large — the combined limit is 15 MB.", "error")
return redirect(url_for("index"))
# ---------------------------------------------------------------- helpers
@app.template_filter("fromjson")
def fromjson_filter(s):
try:
return json.loads(s or "[]")
except Exception:
return []
def slugify(name):
s = re.sub(r"[^a-z0-9]+", "-", name.lower()).strip("-")
return s or "workflow"
def login_required(view):
@functools.wraps(view)
def wrapped(*a, **kw):
if "user_id" not in session:
return redirect(url_for("login", next=request.path))
return view(*a, **kw)
return wrapped
def admin_required(view):
@functools.wraps(view)
def wrapped(*a, **kw):
if "user_id" not in session:
return redirect(url_for("login", next=request.path))
if not session.get("is_admin"):
abort(403)
return view(*a, **kw)
return wrapped
@app.before_request
def open_db():
g.db = db.get_db()
@app.teardown_request
def close_db(exc):
d = g.pop("db", None)
if d:
d.close()
def user_workflows(user_id):
if session.get("is_admin"):
return g.db.execute("SELECT * FROM workflows WHERE active=1 ORDER BY name").fetchall()
return g.db.execute(
"SELECT w.* FROM workflows w JOIN user_workflows uw ON uw.workflow_id=w.id "
"WHERE uw.user_id=? AND w.active=1 ORDER BY w.name", (user_id,)).fetchall()
# ---------------------------------------------------------------- auth
def _start_session(user, next_url=None):
session.clear()
session["user_id"] = user["id"]
session["username"] = user["username"]
session["is_admin"] = bool(user["is_admin"])
return redirect(next_url or url_for("index"))
@app.route("/login", methods=["GET", "POST"])
def login():
if request.method == "POST":
user = g.db.execute("SELECT * FROM users WHERE username=?",
(request.form.get("username", "").strip(),)).fetchone()
if user and check_password_hash(user["password_hash"], request.form.get("password", "")):
if user["mfa_secret"]:
session.clear()
session["mfa_pending"] = user["id"]
session["mfa_next"] = request.args.get("next") or ""
return redirect(url_for("login_mfa"))
return _start_session(user, request.args.get("next"))
flash("Wrong username or password.", "error")
return render_template("login.html")
@app.route("/login/mfa", methods=["GET", "POST"])
def login_mfa():
uid = session.get("mfa_pending")
if not uid:
return redirect(url_for("login"))
user = g.db.execute("SELECT * FROM users WHERE id=?", (uid,)).fetchone()
if user is None or not user["mfa_secret"]:
session.clear()
return redirect(url_for("login"))
if request.method == "POST":
if totp.verify(user["mfa_secret"], request.form.get("code", "")):
return _start_session(user, session.get("mfa_next") or None)
flash("Wrong verification code — try again.", "error")
return render_template("login.html", mfa_stage=True, username=user["username"])
@app.route("/logout")
def logout():
session.clear()
return redirect(url_for("login"))
# ---------------------------------------------------------------- account (MFA)
@app.route("/account", methods=["GET", "POST"])
@login_required
def account():
user = g.db.execute("SELECT * FROM users WHERE id=?", (session["user_id"],)).fetchone()
setup_secret = session.get("mfa_setup")
if request.method == "POST":
act = request.form.get("action")
if act == "profile":
email = request.form.get("email", "").strip().lower()
if email and "@" not in email:
flash("That e-mail address doesn't look valid.", "error")
else:
g.db.execute(
"UPDATE users SET first_name=?, last_name=?, unit_location=?, email=? WHERE id=?",
(request.form.get("first_name", "").strip(),
request.form.get("last_name", "").strip(),
request.form.get("unit_location", "").strip(),
email, user["id"]))
g.db.commit()
flash("Profile saved.", "ok")
return redirect(url_for("account"))
if act == "password":
if not check_password_hash(user["password_hash"], request.form.get("current_password", "")):
flash("Your current password is wrong.", "error")
elif not request.form.get("new_password"):
flash("The new password can't be empty.", "error")
elif request.form.get("new_password") != request.form.get("confirm_password"):
flash("The new passwords don't match.", "error")
else:
g.db.execute("UPDATE users SET password_hash=? WHERE id=?",
(generate_password_hash(request.form["new_password"]), user["id"]))
g.db.commit()
flash("Password changed.", "ok")
return redirect(url_for("account"))
if act == "begin":
session["mfa_setup"] = totp.new_secret()
return redirect(url_for("account"))
if act == "cancel":
session.pop("mfa_setup", None)
return redirect(url_for("account"))
if act == "confirm" and setup_secret:
if totp.verify(setup_secret, request.form.get("code", "")):
g.db.execute("UPDATE users SET mfa_secret=? WHERE id=?",
(setup_secret, user["id"]))
g.db.commit()
session.pop("mfa_setup", None)
flash("Two-factor authentication is now enabled on your account.", "ok")
else:
flash("That code didn't match — scan the QR code and try again.", "error")
return redirect(url_for("account"))
if act == "disable" and user["mfa_secret"]:
if totp.verify(user["mfa_secret"], request.form.get("code", "")):
g.db.execute("UPDATE users SET mfa_secret='' WHERE id=?", (user["id"],))
g.db.commit()
flash("Two-factor authentication disabled.", "ok")
else:
flash("Wrong code — MFA was not disabled.", "error")
return redirect(url_for("account"))
otpauth = totp.otpauth_uri(user["username"], setup_secret) if setup_secret else ""
return render_template("account.html", user=user,
setup_secret=setup_secret, otpauth=otpauth)
# ---------------------------------------------------------------- main page
@app.route("/", methods=["GET", "POST"])
@login_required
def index():
workflows = user_workflows(session["user_id"])
user = g.db.execute("SELECT * FROM users WHERE id=?", (session["user_id"],)).fetchone()
if request.method == "POST":
wf_id = request.form.get("workflow_id", type=int)
wf = next((w for w in workflows if w["id"] == wf_id), None)
if wf is None:
abort(403) # not allowed to use this workflow
requester = request.form.get("request_email", "").strip().lower()
subject = request.form.get("subject", "").strip()
body = request.form.get("body", "").strip()
uploads = [(f, f.read()) for f in request.files.getlist("attachments")
if f and f.filename]
total = sum(len(data) for _, data in uploads)
if not requester or "@" not in requester:
flash("Request's e-mail is required and must be a valid address.", "error")
elif not subject or not body:
flash("Subject and message are both required.", "error")
elif total > db.MAX_ATTACHMENT_BYTES:
flash("The uploaded files are too large — the combined limit is 15 MB.", "error")
else:
rid, _ = db.create_request(g.db, wf["id"], "web", requester, subject, body)
db.log(g.db, rid, "matched", f"Submitted on the web form by {session['username']}")
if uploads:
os.makedirs(db.ATTACH_DIR, exist_ok=True)
entries = []
for i, (f, data) in enumerate(uploads, 1):
name = secure_filename(f.filename) or f"attachment-{i}"
stored = f"{rid}_{i}_{name}"
with open(db.attachment_path(stored), "wb") as fh:
fh.write(data)
entries.append({"stored": stored, "name": name})
g.db.execute("UPDATE requests SET attachment=? WHERE id=?",
(json.dumps(entries), rid))
db.log(g.db, rid, "attachments saved",
", ".join(f"'{e['name']}'" for e in entries) +
f" ({total} bytes total) — included in every approval mail")
g.db.commit()
pipeline.send_approval_request(rid)
pipeline.notify_knowledge(rid, f"New request created by {requester}")
flash(f"Request #{rid} sent to the approver of “{wf['name']}”.", "ok")
return redirect(url_for("index"))
my_requests = g.db.execute(
"SELECT r.*, w.name wf_name FROM requests r JOIN workflows w ON w.id=r.workflow_id "
"WHERE r.requester=? OR r.requester=? ORDER BY r.id DESC LIMIT 15",
(session["username"], (user["email"] or "-").lower())).fetchall()
return render_template("index.html", workflows=workflows, my_requests=my_requests,
user_email=user["email"] or "")
# ---------------------------------------------------------------- decision links (from email)
@app.route("/decision/<token>/<action>", methods=["GET", "POST"])
def decision(token, action):
if action not in ("approve", "deny"):
abort(404)
req = g.db.execute("SELECT r.*, w.name wf_name, w.approvers FROM requests r "
"JOIN workflows w ON w.id=r.workflow_id WHERE r.token=?", (token,)).fetchone()
if req is None:
return render_template("decision.html", stage="notfound"), 404
chain = pipeline.approver_chain(req)
approver = chain[min(req["current_step"], len(chain) - 1)] if chain else ""
if request.method == "POST":
decided_by = request.form.get("decided_by", "").strip() or approver
reason = request.form.get("reason", "").strip()
if action == "deny" and not reason:
return render_template("decision.html", stage="confirm", req=req, action=action,
approver=approver, chain=chain,
error="A reason for denial is required."), 400
status, req2, message = pipeline.decide(token, action, decided_by, reason=reason)
return render_template("decision.html", stage="done", req=req2, message=message,
action=action, status=status)
return render_template("decision.html", stage="confirm", req=req, action=action,
approver=approver, chain=chain)
# ---------------------------------------------------------------- admin: dashboard & requests
@app.route("/admin")
@admin_required
def admin():
rows = g.db.execute(
"SELECT r.*, w.name wf_name FROM requests r JOIN workflows w ON w.id=r.workflow_id "
"ORDER BY r.id DESC LIMIT 200").fetchall()
stats = g.db.execute(
"SELECT status, COUNT(*) c FROM requests GROUP BY status").fetchall()
workflows = g.db.execute("SELECT * FROM workflows ORDER BY name").fetchall()
return render_template("admin.html", rows=rows, workflows=workflows,
stats={s["status"]: s["c"] for s in stats})
@app.route("/admin/request/<int:rid>", methods=["GET", "POST"])
@admin_required
def admin_request(rid):
req = g.db.execute("SELECT r.*, w.name wf_name, w.approvers FROM requests r "
"JOIN workflows w ON w.id=r.workflow_id WHERE r.id=?", (rid,)).fetchone()
if req is None:
abort(404)
if request.method == "POST" and req["status"] == "pending":
action = request.form.get("action")
reason = request.form.get("reason", "").strip()
if action == "deny" and not reason:
flash("A reason for denial is required.", "error")
return redirect(url_for("admin_request", rid=rid))
if action in ("approve", "deny"):
pipeline.decide(req["token"], action, f"{session['username']} (admin)", reason=reason)
return redirect(url_for("admin_request", rid=rid))
req = g.db.execute("SELECT r.*, w.name wf_name, w.approvers FROM requests r "
"JOIN workflows w ON w.id=r.workflow_id WHERE r.id=?", (rid,)).fetchone()
logs = g.db.execute("SELECT * FROM request_log WHERE request_id=? ORDER BY id", (rid,)).fetchall()
participants = json.loads(req["participants"])
chain = pipeline.approver_chain(req)
return render_template("request_detail.html", req=req, logs=logs,
participants=participants, chain=chain,
attachments=pipeline.attachment_list(req))
@app.route("/admin/request/<int:rid>/attachment/<int:idx>")
@admin_required
def request_attachment(rid, idx):
req = g.db.execute("SELECT attachment FROM requests WHERE id=?", (rid,)).fetchone()
if req is None:
abort(404)
entries = pipeline.attachment_list(req)
if idx < 0 or idx >= len(entries):
abort(404)
path = db.attachment_path(entries[idx]["stored"])
if not os.path.exists(path):
abort(404)
return send_file(path, as_attachment=True, download_name=entries[idx]["name"])
@app.route("/admin/request/<int:rid>/send-log", methods=["POST"])
@admin_required
def send_request_log(rid):
req = g.db.execute("SELECT r.*, w.name wf_name, w.approvers, w.knowledge FROM requests r "
"JOIN workflows w ON w.id=r.workflow_id WHERE r.id=?", (rid,)).fetchone()
if req is None:
abort(404)
to = request.form.get("email", "").strip()
if not to or "@" not in to:
flash("A valid e-mail address is required to send the log.", "error")
return redirect(url_for("admin_request", rid=rid))
logs = g.db.execute("SELECT * FROM request_log WHERE request_id=? ORDER BY id", (rid,)).fetchall()
chain = pipeline.approver_chain(req)
participants = json.loads(req["participants"])
lines = [
f"Full log of request #{req['id']} — {req['subject']}",
"=" * 60,
f"Workflow: {req['wf_name']}",
f"Requester: {req['requester']}",
f"Created: {req['created_at']}",
f"Status: {req['status'].upper()}",
]
if req["decided_by"]:
lines.append(f"Final decision: by {req['decided_by']} at {req['decided_at']}")
if req["deny_reason"]:
lines.append(f"Denial reason: {req['deny_reason']}")
lines += [
f"Approval chain: {' -> '.join(chain)}",
f"Knowledge chain: {', '.join(pipeline.knowledge_chain(req)) or '(none)'}",
f"People involved: {', '.join(participants)}",
"",
"Message:",
req["body"],
"",
"Event log:",
"-" * 60,
]
for l in logs:
lines.append(f"{l['at']} | {l['event']}")
if l["detail"]:
lines.append(f"{'':25} {l['detail']}")
body = "\n".join(lines)
ok, detail = send_mail(to, f"[{req['wf_name']}] Full log of request #{req['id']}: {req['subject']}",
body, request_id=rid)
db.log(g.db, rid, "log sent" if ok else "mail error",
f"Full log emailed to {to} by {session['username']} — {detail}")
g.db.commit()
flash(f"Full log sent to {to}." if ok else f"Could not send the log: {detail}",
"ok" if ok else "error")
return redirect(url_for("admin_request", rid=rid))
# ---------------------------------------------------------------- admin: workflows
WF_FIELDS = ("approval_subject", "approval_body", "approved_subject",
"approved_body", "denied_subject", "denied_body")
@app.route("/admin/workflows/new", methods=["GET", "POST"])
@app.route("/admin/workflows/<int:wid>", methods=["GET", "POST"])
@admin_required
def workflow_edit(wid=None):
wf = g.db.execute("SELECT * FROM workflows WHERE id=?", (wid,)).fetchone() if wid else None
if wid and wf is None:
abort(404)
if request.method == "POST":
f = request.form
name = f.get("name", "").strip()
approvers = [a.strip() for a in f.get("approvers", "").splitlines() if a.strip()]
knowledge = [a.strip() for a in f.get("knowledge", "").splitlines() if a.strip()]
if not name or not approvers:
flash("A name and at least one approver are required.", "error")
else:
templates = {k: f.get(k, "").strip() or DEFAULT_TEMPLATES[k] for k in WF_FIELDS}
try:
if wf is None:
g.db.execute(
"INSERT INTO workflows (name, slug, description, approvers, knowledge, active,"
" approval_subject, approval_body, approved_subject, approved_body,"
" denied_subject, denied_body, created_at)"
" VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?)",
(name, slugify(name), f.get("description", "").strip(), json.dumps(approvers),
json.dumps(knowledge), 1 if f.get("active") else 0,
*[templates[k] for k in WF_FIELDS], db.now()))
flash(f"Workflow “{name}” created with {len(approvers)} approval step(s).", "ok")
else:
g.db.execute(
"UPDATE workflows SET name=?, slug=?, description=?, approvers=?, knowledge=?,"
" active=?, approval_subject=?, approval_body=?,"
" approved_subject=?, approved_body=?, denied_subject=?, denied_body=? WHERE id=?",
(name, slugify(name), f.get("description", "").strip(), json.dumps(approvers),
json.dumps(knowledge), 1 if f.get("active") else 0,
*[templates[k] for k in WF_FIELDS], wid))
flash(f"Workflow “{name}” saved.", "ok")
g.db.commit()
return redirect(url_for("admin"))
except Exception as e:
flash(f"Could not save: {e}", "error")
return render_template("workflow_edit.html", wf=wf, defaults=DEFAULT_TEMPLATES,
images=_email_images())
# ---------------------------------------------------------------- admin: mail status
@app.route("/admin/mail")
@admin_required
def mail_status():
rows = g.db.execute("SELECT * FROM mail_log ORDER BY id DESC LIMIT 300").fetchall()
stats = {s["status"]: s["c"] for s in
g.db.execute("SELECT status, COUNT(*) c FROM mail_log GROUP BY status")}
copy_fails = g.db.execute(
"SELECT COUNT(*) c FROM mail_log WHERE sent_copy LIKE 'failed%'").fetchone()["c"]
configured = bool(os.environ.get("MAIL_USER"))
return render_template("mail_status.html", rows=rows, stats=stats,
copy_fails=copy_fails, configured=configured)
# ---------------------------------------------------------------- admin: email images
def _email_images():
os.makedirs(db.EMAIL_IMG_DIR, exist_ok=True)
return sorted(f for f in os.listdir(db.EMAIL_IMG_DIR)
if os.path.splitext(f)[1].lower() in db.EMAIL_IMAGE_TYPES)
@app.route("/admin/email-images/upload", methods=["POST"])
@admin_required
def email_image_upload():
nxt = request.form.get("next") or url_for("admin")
upload = request.files.get("image")
if not upload or not upload.filename:
flash("Choose an image file to upload.", "error")
return redirect(nxt)
name = secure_filename(upload.filename)
ext = os.path.splitext(name)[1].lower()
if ext not in db.EMAIL_IMAGE_TYPES:
flash("Only PNG, JPG and GIF images can be used in emails.", "error")
return redirect(nxt)
data = upload.read()
if len(data) > db.MAX_EMAIL_IMAGE_BYTES:
flash("Email images are limited to 2 MB each.", "error")
return redirect(nxt)
os.makedirs(db.EMAIL_IMG_DIR, exist_ok=True)
if os.path.exists(os.path.join(db.EMAIL_IMG_DIR, name)):
flash(f"An image named “{name}” already exists — delete it first or rename the file.", "error")
return redirect(nxt)
with open(os.path.join(db.EMAIL_IMG_DIR, name), "wb") as fh:
fh.write(data)
flash(f"Image “{name}” uploaded — use it with {{image:{name}}} in the email bodies.", "ok")
return redirect(nxt)
@app.route("/admin/email-images/delete", methods=["POST"])
@admin_required
def email_image_delete():
nxt = request.form.get("next") or url_for("admin")
name = secure_filename(request.form.get("name", ""))
path = os.path.join(db.EMAIL_IMG_DIR, name)
if name and os.path.isfile(path):
os.remove(path)
flash(f"Image “{name}” deleted.", "ok")
return redirect(nxt)
@app.route("/admin/email-images/<name>")
@admin_required
def email_image(name):
name = secure_filename(name)
path = os.path.join(db.EMAIL_IMG_DIR, name)
if not (name and os.path.isfile(path)):
abort(404)
return send_file(path)
# ---------------------------------------------------------------- admin: users & access
@app.route("/admin/users", methods=["GET", "POST"])
@admin_required
def users():
if request.method == "POST":
f = request.form
act = f.get("action")
try:
if act == "create":
if not f.get("username") or not f.get("password"):
flash("Username and password are required.", "error")
else:
g.db.execute(
"INSERT INTO users (username, password_hash, email, is_admin, created_at) VALUES (?,?,?,?,?)",
(f["username"].strip(), generate_password_hash(f["password"]),
f.get("email", "").strip(), 1 if f.get("is_admin") else 0, db.now()))
flash(f"User “{f['username']}” created.", "ok")
elif act == "delete":
uid = f.get("user_id", type=int)
if uid == session["user_id"]:
flash("You can't delete your own account.", "error")
else:
g.db.execute("DELETE FROM users WHERE id=?", (uid,))
flash("User deleted.", "ok")
elif act == "mfa_off":
g.db.execute("UPDATE users SET mfa_secret='' WHERE id=?",
(f.get("user_id", type=int),))
flash("MFA disabled for that user — they can re-enable it from their Account page.", "ok")
elif act == "password":
g.db.execute("UPDATE users SET password_hash=? WHERE id=?",
(generate_password_hash(f["password"]), f.get("user_id", type=int)))
flash("Password updated.", "ok")
elif act == "access":
uid = f.get("user_id", type=int)
g.db.execute("DELETE FROM user_workflows WHERE user_id=?", (uid,))
for wid in request.form.getlist("workflow_ids", type=int):
g.db.execute("INSERT OR IGNORE INTO user_workflows (user_id, workflow_id) VALUES (?,?)",
(uid, wid))
flash("Access updated.", "ok")
g.db.commit()
except Exception as e:
flash(f"Error: {e}", "error")
return redirect(url_for("users"))
all_users = g.db.execute("SELECT * FROM users ORDER BY username").fetchall()
workflows = g.db.execute("SELECT * FROM workflows ORDER BY name").fetchall()
access = {}
for row in g.db.execute("SELECT * FROM user_workflows"):
access.setdefault(row["user_id"], set()).add(row["workflow_id"])
return render_template("users.html", users=all_users, workflows=workflows, access=access)
# ---------------------------------------------------------------- boot
db.init_db()
if __name__ == "__main__":
app.run(host=os.environ.get("HOST", "0.0.0.0"),
port=int(os.environ.get("PORT", "8000")),
debug=os.environ.get("FLASK_DEBUG") == "1")
Reference in New Issue Copy Permalink