580 lines
25 KiB
Python
580 lines
25 KiB
Python
"""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")
|