"""College KT portal with admin, lead and user logins. admin: manage users + everything a lead can do lead : add / edit / delete colleges and their KT user : search and view colleges and their KT """ import datetime import hashlib import hmac import html import json import os import re import secrets import smtplib import sqlite3 import threading import time import urllib.parse import urllib.request from email.message import EmailMessage from email.utils import formataddr from functools import wraps from pathlib import Path from flask import (Flask, abort, flash, g, redirect, render_template, request, session, url_for) from werkzeug.security import check_password_hash, generate_password_hash DB_PATH = Path(__file__).with_name("colleges.db") KEY_PATH = Path(__file__).with_name(".secret_key") GOOGLE_CONFIG_PATH = Path(__file__).with_name("google_oauth.json") DOMAINS_PATH = Path(__file__).with_name("allowed_domains.json") SMTP_CONFIG_PATH = Path(__file__).with_name("smtp_config.json") APP_NAME = "College KT Portal" DEFAULT_DOMAINS = ("gmail.com",) TITLES = ("Dr.", "Mr.", "Ms.", "Mrs.", "Prof.") KINDS = ( ("short", "Short answer"), ("paragraph", "Paragraph"), ("multiple_choice", "Multiple choice"), ("checkboxes", "Checkboxes"), ("dropdown", "Drop-down"), ("date", "Date"), ("time", "Time"), ) KIND_LABELS = dict(KINDS) CHOICE_KINDS = ("multiple_choice", "checkboxes", "dropdown") FACULTY_EMAIL_RE = re.compile(r"[^@\s]+@[^@\s]+\.[^@\s]+") EMAIL_RE = re.compile(r"[a-z0-9._+-]+@([a-z0-9-]+(?:\.[a-z0-9-]+)+)") ROLES = ("admin", "lead", "user") MANAGED_ROLES = ("lead", "user") # roles an admin may create or assign app = Flask(__name__) if not KEY_PATH.exists(): KEY_PATH.write_text(secrets.token_hex(32)) app.secret_key = KEY_PATH.read_text() def db(): if "db" not in g: g.db = sqlite3.connect(DB_PATH) g.db.row_factory = sqlite3.Row g.db.execute("PRAGMA foreign_keys = ON") return g.db @app.teardown_appcontext def close_db(_): conn = g.pop("db", None) if conn: conn.close() def init_db(): conn = sqlite3.connect(DB_PATH) conn.executescript( """ CREATE TABLE IF NOT EXISTS colleges ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE COLLATE NOCASE ); CREATE TABLE IF NOT EXISTS kts ( id INTEGER PRIMARY KEY AUTOINCREMENT, college_id INTEGER NOT NULL REFERENCES colleges(id) ON DELETE CASCADE, title TEXT NOT NULL DEFAULT '', text TEXT NOT NULL, kind TEXT NOT NULL DEFAULT 'paragraph', options TEXT, faculty_id INTEGER REFERENCES faculty(id) ON DELETE SET NULL ); CREATE TABLE IF NOT EXISTS faculty ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL CHECK (title IN ('Dr.', 'Mr.', 'Ms.', 'Mrs.', 'Prof.')), first_name TEXT NOT NULL, last_name TEXT NOT NULL, email TEXT NOT NULL UNIQUE COLLATE NOCASE ); CREATE TABLE IF NOT EXISTS login_otps ( email TEXT PRIMARY KEY COLLATE NOCASE, code_hash TEXT NOT NULL, expires_at INTEGER NOT NULL, attempts INTEGER NOT NULL DEFAULT 0 ); CREATE TABLE IF NOT EXISTS otp_log ( id INTEGER PRIMARY KEY AUTOINCREMENT, email TEXT NOT NULL, ip TEXT NOT NULL, ts INTEGER NOT NULL ); CREATE TABLE IF NOT EXISTS settings ( key TEXT PRIMARY KEY, value TEXT NOT NULL ); CREATE TABLE IF NOT EXISTS users ( id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL UNIQUE COLLATE NOCASE, password_hash TEXT NOT NULL, role TEXT NOT NULL CHECK (role IN ('admin', 'lead', 'user')), email TEXT ); """ ) if "email" not in [r[1] for r in conn.execute("PRAGMA table_info(users)")]: conn.execute("ALTER TABLE users ADD COLUMN email TEXT") conn.execute("CREATE UNIQUE INDEX IF NOT EXISTS users_email ON users(email COLLATE NOCASE)") # migrate the old single-table layout (colleges.kt) into separate KT rows cols = [r[1] for r in conn.execute("PRAGMA table_info(colleges)")] if "kt" in cols: conn.executescript( """ INSERT INTO kts (college_id, text) SELECT id, kt FROM colleges WHERE trim(kt) <> ''; CREATE TABLE colleges_new ( id INTEGER PRIMARY KEY AUTOINCREMENT, name TEXT NOT NULL UNIQUE COLLATE NOCASE ); INSERT INTO colleges_new (id, name) SELECT id, name FROM colleges; DROP TABLE colleges; ALTER TABLE colleges_new RENAME TO colleges; """ ) if "title" not in [r[1] for r in conn.execute("PRAGMA table_info(kts)")]: conn.execute("ALTER TABLE kts ADD COLUMN title TEXT NOT NULL DEFAULT ''") if "faculty_id" not in [r[1] for r in conn.execute("PRAGMA table_info(kts)")]: conn.execute("ALTER TABLE kts ADD COLUMN faculty_id INTEGER REFERENCES faculty(id) ON DELETE SET NULL") if "name" not in [r[1] for r in conn.execute("PRAGMA table_info(users)")]: conn.execute("ALTER TABLE users ADD COLUMN name TEXT") for table in ("colleges", "faculty", "kts"): # who added each record (id follows renames, name survives deletion) have = [r[1] for r in conn.execute(f"PRAGMA table_info({table})")] if "created_by" not in have: conn.execute(f"ALTER TABLE {table} ADD COLUMN created_by INTEGER REFERENCES users(id) ON DELETE SET NULL") if "created_by_name" not in have: conn.execute(f"ALTER TABLE {table} ADD COLUMN created_by_name TEXT") kt_cols = [r[1] for r in conn.execute("PRAGMA table_info(kts)")] if "kind" not in kt_cols: conn.execute("ALTER TABLE kts ADD COLUMN kind TEXT NOT NULL DEFAULT 'paragraph'") if "options" not in kt_cols: conn.execute("ALTER TABLE kts ADD COLUMN options TEXT") if not conn.execute("SELECT 1 FROM users").fetchone(): conn.execute( "INSERT INTO users (username, password_hash, role) VALUES (?, ?, 'admin')", ("admin", generate_password_hash("admin123")), ) print("Created default admin login -> username: admin password: admin123 (change it!)") conn.commit() conn.close() # ---- auth helpers --------------------------------------------------------- @app.before_request def load_user(): g.user = None uid = session.get("uid") if uid: g.user = db().execute("SELECT * FROM users WHERE id = ?", (uid,)).fetchone() if g.user is None: session.clear() if request.method == "POST": token = session.get("csrf") if not token or token != request.form.get("csrf"): abort(400, "Invalid CSRF token") @app.context_processor def inject_csrf(): if "csrf" not in session: session["csrf"] = secrets.token_hex(16) return {"csrf_token": session["csrf"], "user": g.get("user")} def roles_required(*roles): def deco(fn): @wraps(fn) def wrapper(*a, **kw): if g.user is None: return redirect(url_for("login", next=request.path)) if g.user["role"] not in roles: abort(403) return fn(*a, **kw) return wrapper return deco def _me(): """(id, display name) of the signed-in person, stored with every record they add.""" return g.user["id"], (g.user["name"] or g.user["username"]) def _added_by(table, user): """SQL for the adder's current name; falls back to the saved name if that login was deleted.""" return f"COALESCE(NULLIF({user}.name, ''), {user}.username, {table}.created_by_name)" ANY = ROLES EDITORS = ("admin", "lead") # ---- email ---------------------------------------------------------------- DEFAULT_SUBJECT = "Welcome to {{app_name}}" DEFAULT_BODY = """

Welcome to {{app_name}}

Hi {{name}},

{{added_by}} has given you access to {{app_name}} as a {{role}}.

Sign in with your Google account ({{email}}). You do not need a separate password.

Sign in

If the button does not work, open this link: {{login_url}}

""" TEMPLATE_VARS = ("app_name", "name", "email", "role", "login_url", "added_by") _VAR_RE = re.compile(r"\{\{\s*(" + "|".join(TEMPLATE_VARS) + r")\s*\}\}") class MailError(Exception): pass def get_setting(key, default): row = db().execute("SELECT value FROM settings WHERE key = ?", (key,)).fetchone() return row["value"] if row else default def set_setting(key, value): db().execute("INSERT INTO settings (key, value) VALUES (?, ?) " "ON CONFLICT(key) DO UPDATE SET value = excluded.value", (key, value)) def fill_template(text, values, escape): """Replace {{placeholders}}. Values are HTML-escaped in the body, and forced onto one line in the subject.""" def sub(m): v = str(values.get(m.group(1), "")) return html.escape(v) if escape else " ".join(v.split()) return _VAR_RE.sub(sub, text) def html_to_text(markup): """Plain-text version of the email for clients that do not show HTML.""" t = re.sub(r"(?is)<(style|script)\b.*?", "", markup) t = re.sub(r'(?is)]*href=["\']([^"\']+)["\'][^>]*>(.*?)', r"\2 (\1)", t) t = re.sub(r"(?i)|", "\n", t) t = html.unescape(re.sub(r"<[^>]+>", "", t)) return re.sub(r"\n\s*\n\s*\n+", "\n\n", "\n".join(line.strip() for line in t.splitlines())).strip() def smtp_config(): """Mail server settings from smtp_config.json or SMTP_* environment variables; None if not set up.""" cfg = {} if SMTP_CONFIG_PATH.exists(): try: cfg = json.loads(SMTP_CONFIG_PATH.read_text(encoding="utf-8")) except ValueError: cfg = {} for key in ("host", "port", "username", "password", "from_email", "from_name", "use_tls", "public_url"): if os.environ.get("SMTP_" + key.upper()): cfg[key] = os.environ["SMTP_" + key.upper()] if not cfg.get("host") or not cfg.get("from_email"): return None try: cfg["port"] = int(cfg.get("port") or 587) except ValueError: cfg["port"] = 587 cfg["use_tls"] = str(cfg.get("use_tls", True)).lower() not in ("0", "false", "no") return cfg def send_mail(to, subject, body_html): cfg = smtp_config() if not cfg: raise MailError("email sending is not set up yet (see README)") msg = EmailMessage() msg["Subject"] = " ".join(subject.split())[:200] msg["From"] = formataddr((cfg.get("from_name") or "", cfg["from_email"])) msg["To"] = to msg.set_content(html_to_text(body_html)) msg.add_alternative(body_html, subtype="html") try: server = (smtplib.SMTP_SSL if cfg["port"] == 465 else smtplib.SMTP)(cfg["host"], cfg["port"], timeout=15) with server: if cfg["port"] != 465 and cfg["use_tls"]: server.starttls() if cfg.get("username"): server.login(cfg["username"], cfg.get("password", "")) server.send_message(msg) except smtplib.SMTPAuthenticationError: raise MailError("the mail server rejected the login (check the username and password)") except (smtplib.SMTPException, OSError) as e: raise MailError(f"could not reach the mail server ({e.__class__.__name__})") def portal_login_url(role=None): """The sign-in address: /admin/login for the admin, /login (shared) for Leads and Users.""" path = "/admin/login" if role == "admin" else "/login" base = (smtp_config() or {}).get("public_url") return base.rstrip("/") + path if base else request.host_url.rstrip("/") + path def send_welcome(name, email, role, subject_prefix=""): """Email the saved welcome template to a new Lead/User (values are filled in and escaped).""" values = {"app_name": APP_NAME, "name": name or email, "email": email, "role": role.capitalize(), "login_url": portal_login_url(role), "added_by": _me()[1]} subject = fill_template(get_setting("email_subject", DEFAULT_SUBJECT), values, False) send_mail(email, subject_prefix + subject, fill_template(get_setting("email_body", DEFAULT_BODY), values, True)) # ---- login ---------------------------------------------------------------- def allowed_domains(): """Email domains that may sign in. Read on every call, so edits to allowed_domains.json apply at once.""" try: domains = json.loads(DOMAINS_PATH.read_text(encoding="utf-8")) return [d.strip().lower() for d in domains if isinstance(d, str) and d.strip()] except (OSError, ValueError): return list(DEFAULT_DOMAINS) def valid_email(email): """True only for name@domain where domain is exactly one of the allowed domains (no subdomains).""" m = EMAIL_RE.fullmatch(email) return bool(m) and m.group(1) in allowed_domains() def google_config(): cid, secret = os.environ.get("GOOGLE_CLIENT_ID"), os.environ.get("GOOGLE_CLIENT_SECRET") if not (cid and secret) and GOOGLE_CONFIG_PATH.exists(): cfg = json.loads(GOOGLE_CONFIG_PATH.read_text(encoding="utf-8")) cid, secret = cfg.get("client_id"), cfg.get("client_secret") return (cid, secret) if cid and secret else (None, None) def google_userinfo(code, redirect_uri): """Exchange the authorization code with Google and return the signed-in account's profile.""" cid, secret = google_config() body = urllib.parse.urlencode({ "code": code, "client_id": cid, "client_secret": secret, "redirect_uri": redirect_uri, "grant_type": "authorization_code", }).encode() with urllib.request.urlopen("https://oauth2.googleapis.com/token", body, timeout=10) as r: token = json.load(r)["access_token"] req = urllib.request.Request( "https://openidconnect.googleapis.com/v1/userinfo", headers={"Authorization": f"Bearer {token}"}) with urllib.request.urlopen(req, timeout=10) as r: return json.load(r) ROLE_LOGIN = {"admin": "admin_login", "lead": "login", "user": "login"} # where each role signs in / returns to @app.route("/login") def login(): """Leads and Users sign in here (Google or an emailed code). The admin has a separate address.""" if g.user: return redirect(url_for("search")) return render_template("login_member.html", google_ready=google_config()[0] is not None, otp_on=otp_enabled(), mail_ready=smtp_config() is not None) @app.route("/lead/login") @app.route("/user/login") def old_member_login(): """Earlier per-role addresses (already shared in emails) now forward to the shared page.""" return redirect(url_for("login")) @app.route("/admin/login", methods=["GET", "POST"]) def admin_login(): """The admin signs in with a password.""" if g.user: return redirect(url_for("search")) if request.method == "POST": row = db().execute( "SELECT * FROM users WHERE username = ? AND role = 'admin'", (request.form["username"].strip(),) ).fetchone() if row and check_password_hash(row["password_hash"], request.form["password"]): session.clear() session["uid"] = row["id"] nxt = request.args.get("next", "") return redirect(nxt if nxt.startswith("/") and not nxt.startswith("//") else url_for("search")) flash("Invalid admin username or password.", "error") return render_template("login_admin.html") # ---- sign in with an emailed one-time code (Leads and Users) ------------------------------ OTP_TTL = 10 * 60 # a code works for 10 minutes OTP_MAX_ATTEMPTS = 5 # wrong guesses before the code is thrown away OTP_COOLDOWN = 60 # seconds between code requests for one email OTP_PER_EMAIL_HOUR = 5 OTP_PER_IP_HOUR = 20 app.config.setdefault("MAIL_ASYNC", True) # send in the background so replies take the same time for every address def otp_enabled(): return get_setting("otp_enabled", "1") == "1" def _otp_hash(email, code): return hmac.new(app.secret_key.encode(), f"{email}:{code}".encode(), hashlib.sha256).hexdigest() def _otp_email(code): subject = f"Your {APP_NAME} sign-in code" body = ( '
' f'

{APP_NAME}

' "

Use this code to sign in:

" f'

{code}

' f"

It expires in {OTP_TTL // 60} minutes and can be used once.

" '

If you did not try to sign in, you can ignore this email.

' "
") return subject, body def _deliver(to, subject, body): def job(): try: send_mail(to, subject, body) except MailError as e: app.logger.warning("Sign-in code email to %s failed: %s", to, e) if app.config["MAIL_ASYNC"]: threading.Thread(target=job, daemon=True).start() else: job() def _request_otp(email): """Create and email a code if the address belongs to a Lead or User. Returns 'ok', 'wait' or 'limit'. The answer never depends on whether the address is registered.""" now, ip, d = int(time.time()), request.remote_addr or "?", db() d.execute("DELETE FROM otp_log WHERE ts < ?", (now - 3600,)) d.execute("DELETE FROM login_otps WHERE expires_at < ?", (now,)) last = d.execute("SELECT max(ts) FROM otp_log WHERE email = ?", (email,)).fetchone()[0] if last and now - last < OTP_COOLDOWN: return "wait" if (d.execute("SELECT count(*) FROM otp_log WHERE email = ?", (email,)).fetchone()[0] >= OTP_PER_EMAIL_HOUR or d.execute("SELECT count(*) FROM otp_log WHERE ip = ?", (ip,)).fetchone()[0] >= OTP_PER_IP_HOUR): return "limit" d.execute("INSERT INTO otp_log (email, ip, ts) VALUES (?, ?, ?)", (email, ip, now)) person = d.execute("SELECT 1 FROM users WHERE email = ? AND role IN ('lead', 'user')", (email,)).fetchone() mail = None if person and valid_email(email): code = f"{secrets.randbelow(10**6):06d}" # one active code per address; a new one replaces the old d.execute("INSERT INTO login_otps (email, code_hash, expires_at, attempts) VALUES (?, ?, ?, 0) " "ON CONFLICT(email) DO UPDATE SET code_hash = excluded.code_hash, " "expires_at = excluded.expires_at, attempts = 0", (email, _otp_hash(email, code), now + OTP_TTL)) mail = _otp_email(code) d.commit() if mail: _deliver(email, *mail) return "ok" def _send_code(email): """Shared by 'send' and 'resend': request a code and tell the person what happened.""" if not otp_enabled(): flash("Sign-in with an email code is turned off.", "error") return False if smtp_config() is None: flash("Email sign-in is not available yet because email sending is not set up. Ask the admin.", "error") return False status = _request_otp(email) if status == "wait": flash("A code was just sent. Please wait about a minute before asking for another one.", "error") return True # they already have a code waiting, so go straight to the code page if status == "limit": flash("Too many code requests. Please try again later.", "error") return False flash("If that email is registered, a 6-digit code has been sent to it. It expires in " f"{OTP_TTL // 60} minutes.", "ok") return True @app.post("/login/otp/send") def otp_send(): email = request.form.get("email", "").strip().lower() if len(email) > 254 or not FACULTY_EMAIL_RE.fullmatch(email): flash("Enter a valid email address.", "error") return redirect(url_for("login")) if _send_code(email): session["otp_email"] = email return redirect(url_for("otp_page")) return redirect(url_for("login")) @app.post("/login/otp/resend") def otp_resend(): email = session.get("otp_email") if not email: return redirect(url_for("login")) _send_code(email) return redirect(url_for("otp_page")) @app.route("/login/otp") def otp_page(): email = session.get("otp_email") if not email or not otp_enabled(): return redirect(url_for("login")) return render_template("otp.html", email=email, minutes=OTP_TTL // 60) @app.route("/login/otp/change") def otp_change(): session.pop("otp_email", None) return redirect(url_for("login")) @app.post("/login/otp/verify") def otp_verify(): email = session.get("otp_email") if not email or not otp_enabled(): return redirect(url_for("login")) code = re.sub(r"\D", "", request.form.get("code", "")) d, now = db(), int(time.time()) row = d.execute("SELECT * FROM login_otps WHERE email = ?", (email,)).fetchone() person = d.execute("SELECT * FROM users WHERE email = ? AND role IN ('lead', 'user')", (email,)).fetchone() ok = bool(row) and row["expires_at"] >= now and row["attempts"] < OTP_MAX_ATTEMPTS if ok and hmac.compare_digest(row["code_hash"], _otp_hash(email, code)) and person and valid_email(email): d.execute("DELETE FROM login_otps WHERE email = ?", (email,)) # single use d.commit() session.clear() session["uid"] = person["id"] return redirect(url_for("search")) if row: if row["attempts"] + 1 >= OTP_MAX_ATTEMPTS or row["expires_at"] < now: d.execute("DELETE FROM login_otps WHERE email = ?", (email,)) else: d.execute("UPDATE login_otps SET attempts = attempts + 1 WHERE email = ?", (email,)) d.commit() flash("That code is not valid or has expired. Check the code or ask for a new one.", "error") return redirect(url_for("otp_page")) @app.route("/auth/google") def google_login(): cid, _ = google_config() if not cid: flash("Google sign-in is not configured yet. Ask the admin.", "error") return redirect(url_for("login")) session["oauth_state"] = secrets.token_urlsafe(16) params = { "client_id": cid, "redirect_uri": url_for("google_callback", _external=True), "response_type": "code", "scope": "openid email profile", "state": session["oauth_state"], "prompt": "select_account", } return redirect("https://accounts.google.com/o/oauth2/v2/auth?" + urllib.parse.urlencode(params)) @app.route("/auth/google/callback") def google_callback(): state = session.pop("oauth_state", None) session.pop("login_role", None) # left over from the earlier per-role pages code = request.args.get("code") if not state or request.args.get("state") != state or not code: flash("Google sign-in was cancelled or failed. Please try again.", "error") return redirect(url_for("login")) try: info = google_userinfo(code, url_for("google_callback", _external=True)) except Exception: flash("Could not complete Google sign-in. Please try again.", "error") return redirect(url_for("login")) email = str(info.get("email", "")).lower() row = db().execute("SELECT * FROM users WHERE email = ? AND role IN ('lead', 'user')", (email,)).fetchone() if not info.get("email_verified") or not valid_email(email) or row is None: flash("This email has not been given access. Ask the admin to add it.", "error") return redirect(url_for("login")) if not row["name"] and str(info.get("name", "")).strip(): db().execute("UPDATE users SET name = ? WHERE id = ?", (str(info["name"]).strip()[:100], row["id"])) db().commit() session.clear() session["uid"] = row["id"] return redirect(url_for("search")) @app.post("/logout") def logout(): role = g.user["role"] if g.user else None session.clear() return redirect(url_for(ROLE_LOGIN.get(role, "login"))) # ---- search KT ------------------------------------------------------------ @app.route("/") def index(): return redirect(url_for("search")) def _kt_view(row): """A KT row plus what the template needs to show it according to its description type.""" v = dict(row) v["kind_label"] = KIND_LABELS.get(v.get("kind"), "Paragraph") v["choices"] = [] if v.get("kind") in CHOICE_KINDS and v.get("options"): try: data = json.loads(v["options"]) chosen = set(data.get("selected", [])) v["choices"] = [(o, i in chosen) for i, o in enumerate(data.get("options", []))] except (ValueError, AttributeError): pass if v.get("kind") == "date": try: v["text"] = datetime.date.fromisoformat(v["text"]).strftime("%d %b %Y") except ValueError: pass return v @app.route("/search") @roles_required(*ANY) def search(): """No query/selection: list of colleges only. Query or ?college=id: matching colleges with their KT.""" q = request.args.get("q", "").strip() cid = request.args.get("college", type=int) names = [r["name"] for r in db().execute("SELECT name FROM colleges ORDER BY name")] kts = {} if cid: colleges = db().execute("SELECT * FROM colleges WHERE id = ?", (cid,)).fetchall() where, params = "college_id = ?", (cid,) elif q: colleges = db().execute( "SELECT * FROM colleges WHERE name LIKE ? ORDER BY name", (f"%{q}%",) ).fetchall() where, params = "college_id IN (SELECT id FROM colleges WHERE name LIKE ?)", (f"%{q}%",) else: colleges = db().execute( "SELECT c.*, (SELECT count(*) FROM kts WHERE college_id = c.id) AS kt_count, " + _added_by("c", "cu") + " AS added_by " "FROM colleges c LEFT JOIN users cu ON cu.id = c.created_by ORDER BY c.name" ).fetchall() return render_template("search.html", colleges=colleges, kts=kts, q=q, searched=False, names=names) for k in db().execute( "SELECT kts.*, f.title AS f_title, f.first_name AS f_first, f.last_name AS f_last, " + _added_by("kts", "ku") + " AS added_by " f"FROM kts LEFT JOIN faculty f ON f.id = kts.faculty_id LEFT JOIN users ku ON ku.id = kts.created_by " f"WHERE {where} ORDER BY kts.id", params): kts.setdefault(k["college_id"], []).append(_kt_view(k)) return render_template("search.html", colleges=colleges, kts=kts, q=q, searched=True, names=names) # ---- add / manage colleges ------------------------------------------------ def _college_rows(): return db().execute( "SELECT c.*, " + _added_by("c", "cu") + " AS added_by " "FROM colleges c LEFT JOIN users cu ON cu.id = c.created_by ORDER BY c.name" ).fetchall() @app.route("/colleges/new", methods=["GET", "POST"]) @roles_required(*EDITORS) def college_new(): if request.method == "POST": name = request.form["name"].strip() if not name: flash("College name is required.", "error") else: try: db().execute("INSERT INTO colleges (name, created_by, created_by_name) VALUES (?, ?, ?)", (name, *_me())) db().commit() flash(f"College {name} added. You can now add its KT.", "ok") return redirect(url_for("college_new")) except sqlite3.IntegrityError: flash("A college with that name already exists.", "error") return render_template("college_new.html", rows=_college_rows()) @app.route("/colleges//edit", methods=["GET", "POST"]) @roles_required(*EDITORS) def college_edit(cid): row = db().execute("SELECT * FROM colleges WHERE id = ?", (cid,)).fetchone() if row is None: abort(404) if request.method == "POST": name = request.form["name"].strip() if not name: flash("College name is required.", "error") else: try: db().execute("UPDATE colleges SET name = ? WHERE id = ?", (name, cid)) db().commit() flash("College renamed.", "ok") return redirect(url_for("college_new")) except sqlite3.IntegrityError: flash("A college with that name already exists.", "error") return render_template("college_edit.html", college=row) @app.post("/colleges//delete") @roles_required(*EDITORS) def college_delete(cid): db().execute("DELETE FROM colleges WHERE id = ?", (cid,)) db().commit() flash("College and its KT deleted.", "ok") return redirect(url_for("college_new")) # ---- add / edit KT -------------------------------------------------------- def _faculty_rows(): return db().execute("SELECT * FROM faculty ORDER BY first_name, last_name").fetchall() def _parse_desc(kind, payload): """Validate one KT description. Returns (text, options_json, is_blank, error).""" if kind not in KIND_LABELS: return "", None, False, "choose a valid description type." if kind not in CHOICE_KINDS: value = str(payload.get("value", "")).strip() if kind == "short": value = " ".join(value.split())[:500] value = value[:10000] if not value: return "", None, True, "enter the description." try: if kind == "date": datetime.date.fromisoformat(value) elif kind == "time": datetime.time.fromisoformat(value) except ValueError: return value, None, False, f"enter a valid {kind}." return value, None, False, None raw = payload.get("options") raw = raw if isinstance(raw, list) else [] chosen_raw = payload.get("selected") chosen_raw = chosen_raw if isinstance(chosen_raw, list) else [] options, remap = [], {} for i, o in enumerate(raw[:50]): o = str(o).strip()[:200] if o: remap[i] = len(options) options.append(o) if not options: return "", None, True, "add the options." if len({o.lower() for o in options}) != len(options): return "", None, False, "the options must all be different." selected = sorted({remap[i] for i in chosen_raw if type(i) is int and i in remap}) if kind == "checkboxes": if not selected: return "", None, False, "tick at least one option." else: if len(options) < 2: return "", None, False, "add at least two options." if len(selected) != 1: return "", None, False, "select one option." text = ", ".join(options[i] for i in selected) return text, json.dumps({"options": options, "selected": selected}), False, None def _read_kt_rows(): """One dict per KT fieldset. All field lists stay aligned because each fieldset posts every field once.""" titles = [x.strip() for x in request.form.getlist("title")] kinds = request.form.getlist("kind") payloads = request.form.getlist("payload") flags = request.form.getlist("use_faculty") facs = request.form.getlist("faculty") rows = [] for i in range(min(len(titles), len(kinds), len(payloads))): try: payload = json.loads(payloads[i]) except ValueError: payload = {} if not isinstance(payload, dict): payload = {} text, options, blank, error = _parse_desc(kinds[i], payload) rows.append({ "title": titles[i], "kind": kinds[i] if kinds[i] in KIND_LABELS else "paragraph", "payload": payload, "text": text, "options": options, "blank": blank, "error": error, "use": i < len(flags) and flags[i] == "1", "fac": int(facs[i]) if i < len(facs) and facs[i].isdigit() else None, }) return rows def _faculty_error(rows): """Every KT whose toggle is on must point at an existing faculty.""" ids = {r["id"] for r in db().execute("SELECT id FROM faculty")} for n, r in enumerate(rows, 1): if r["use"] and r["fac"] not in ids: return f"KT {n}: choose a faculty, or switch its toggle off." return None def _rows_error(rows): for n, r in enumerate(rows, 1): if not r["title"]: return f"KT {n}: enter a title." if r["error"]: return f"KT {n}: {r['error']}" return _faculty_error(rows) BLANK_ROW = {"title": "", "kind": "paragraph", "payload": {"value": ""}, "use": False, "fac": None} @app.route("/kt/new", methods=["GET", "POST"]) @roles_required(*EDITORS) def kt_new(): selected = request.values.get("college", type=int) items = [BLANK_ROW] if request.method == "POST": rows = _read_kt_rows() items = [r for r in rows if r["title"] or not r["blank"]] # ignore fully blank fieldsets college = db().execute("SELECT name FROM colleges WHERE id = ?", (selected,)).fetchone() if not college: items = items or [BLANK_ROW] flash("Please choose a college.", "error") elif not items: items = [BLANK_ROW] flash("Enter a KT title and description.", "error") elif _rows_error(items): flash(_rows_error(items), "error") else: db().executemany( "INSERT INTO kts (college_id, title, kind, text, options, faculty_id, created_by, created_by_name) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", [(selected, r["title"], r["kind"], r["text"], r["options"], r["fac"] if r["use"] else None, *_me()) for r in items], ) db().commit() flash(f"{len(items)} KT saved.", "ok") return redirect(url_for("search", q=college["name"])) return render_template("kt_form.html", colleges=_college_rows(), selected=selected, kt=None, items=items, faculty=_faculty_rows(), kinds=KINDS) @app.route("/kt//edit", methods=["GET", "POST"]) @roles_required(*EDITORS) def kt_edit(kid): kt = db().execute( "SELECT kts.*, colleges.name AS college FROM kts JOIN colleges ON colleges.id = kts.college_id " "WHERE kts.id = ?", (kid,)).fetchone() if kt is None: abort(404) payload = {"value": kt["text"]} if kt["kind"] in CHOICE_KINDS and kt["options"]: try: payload = json.loads(kt["options"]) except ValueError: pass items = [{"title": kt["title"], "kind": kt["kind"], "payload": payload, "use": kt["faculty_id"] is not None, "fac": kt["faculty_id"]}] if request.method == "POST": items = _read_kt_rows()[:1] or items if _rows_error(items): flash(_rows_error(items), "error") else: r = items[0] db().execute( "UPDATE kts SET title = ?, kind = ?, text = ?, options = ?, faculty_id = ? WHERE id = ?", (r["title"], r["kind"], r["text"], r["options"], r["fac"] if r["use"] else None, kid)) db().commit() flash("KT updated.", "ok") return redirect(url_for("search", q=kt["college"])) return render_template("kt_form.html", colleges=[], selected=kt["college_id"], kt=kt, items=items, faculty=_faculty_rows(), kinds=KINDS) @app.post("/kt//delete") @roles_required(*EDITORS) def kt_delete(kid): db().execute("DELETE FROM kts WHERE id = ?", (kid,)) db().commit() flash("KT deleted.", "ok") return redirect(url_for("search")) # ---- faculty -------------------------------------------------------------- def _faculty_form(): """Read and validate the faculty form; returns (values, error message or None).""" f = {k: request.form.get(k, "").strip() for k in ("title", "first_name", "last_name", "email")} f["email"] = f["email"].lower() error = None if f["title"] not in TITLES: error = "Please choose a title." elif not f["first_name"] or not f["last_name"]: error = "First name and last name are required." elif not FACULTY_EMAIL_RE.fullmatch(f["email"]): error = "Enter a valid Mail ID." return f, error @app.route("/faculty/new", methods=["GET", "POST"]) @roles_required(*EDITORS) def faculty_new(): values = {} if request.method == "POST": values, error = _faculty_form() if error: flash(error, "error") else: try: db().execute( "INSERT INTO faculty (title, first_name, last_name, email, created_by, created_by_name) " "VALUES (?, ?, ?, ?, ?, ?)", (values["title"], values["first_name"], values["last_name"], values["email"], *_me()), ) db().commit() flash(f"{values['title']} {values['first_name']} {values['last_name']} added.", "ok") return redirect(url_for("faculty_new")) except sqlite3.IntegrityError: flash("A faculty member with that Mail ID already exists.", "error") rows = db().execute( "SELECT p.*, " + _added_by("p", "pu") + " AS added_by " "FROM faculty p LEFT JOIN users pu ON pu.id = p.created_by ORDER BY p.first_name, p.last_name" ).fetchall() return render_template("faculty_new.html", rows=rows, titles=TITLES, f=values) @app.route("/faculty//edit", methods=["GET", "POST"]) @roles_required(*EDITORS) def faculty_edit(fid): row = db().execute("SELECT * FROM faculty WHERE id = ?", (fid,)).fetchone() if row is None: abort(404) values = dict(row) if request.method == "POST": values, error = _faculty_form() if error: flash(error, "error") else: try: db().execute( "UPDATE faculty SET title = ?, first_name = ?, last_name = ?, email = ? WHERE id = ?", (values["title"], values["first_name"], values["last_name"], values["email"], fid), ) db().commit() flash("Faculty updated.", "ok") return redirect(url_for("faculty_new")) except sqlite3.IntegrityError: flash("A faculty member with that Mail ID already exists.", "error") return render_template("faculty_edit.html", titles=TITLES, f=values, fid=fid) @app.post("/faculty//delete") @roles_required(*EDITORS) def faculty_delete(fid): db().execute("DELETE FROM faculty WHERE id = ?", (fid,)) db().commit() flash("Faculty deleted.", "ok") return redirect(url_for("faculty_new")) # ---- user management (admin) --------------------------------------------- def _clean_email(value): email = value.strip().lower() return email if valid_email(email) else None @app.route("/users", methods=["GET", "POST"]) @roles_required("admin") def users(): if request.method == "POST": email = _clean_email(request.form["email"]) role = request.form["role"] name = request.form.get("name", "").strip()[:100] if role not in MANAGED_ROLES: flash("You can only add Lead or User logins.", "error") elif not name: flash("Enter the person's name.", "error") elif not email: flash("Enter a valid email from an allowed domain: " + ", ".join(allowed_domains()) + ".", "error") else: try: # no password: these logins sign in with Google only db().execute( "INSERT INTO users (username, password_hash, role, email, name) VALUES (?, '', ?, ?, ?)", (email, role, email, name), ) db().commit() flash(f"{email} added as {role}. They can now sign in with Google.", "ok") if get_setting("email_enabled", "1") == "1": try: send_welcome(name, email, role) flash(f"Welcome email sent to {email}.", "ok") except MailError as e: flash(f"The welcome email was not sent: {e}. Use \"Welcome email\" on that row to resend it.", "error") except sqlite3.IntegrityError: flash("That email is already added.", "error") return redirect(url_for("users")) rows = db().execute("SELECT id, username, role, email, name FROM users ORDER BY role, username").fetchall() return render_template("users.html", rows=rows, roles=MANAGED_ROLES, google_ready=google_config()[0] is not None, domains=allowed_domains(), mail_ready=smtp_config() is not None, mail_on=get_setting("email_enabled", "1") == "1", signin_url=portal_login_url("lead")) @app.post("/users//update") @roles_required("admin") def user_update(uid): target = db().execute("SELECT * FROM users WHERE id = ?", (uid,)).fetchone() if target is None: abort(404) name = request.form.get("name", "").strip()[:100] if target["role"] == "admin": # the admin has a name and a password, but no role or email to change password = request.form.get("password", "") if password and len(password) < 6: flash("Password must be 6+ characters.", "error") return redirect(url_for("users")) if name: db().execute("UPDATE users SET name = ? WHERE id = ?", (name, uid)) if password: db().execute("UPDATE users SET password_hash = ? WHERE id = ?", (generate_password_hash(password), uid)) db().commit() flash("Admin updated.", "ok") return redirect(url_for("users")) role = request.form.get("role") email = _clean_email(request.form.get("email", "")) if role not in MANAGED_ROLES: flash("Role must be Lead or User.", "error") elif not email: flash("Enter a valid email from an allowed domain: " + ", ".join(allowed_domains()) + ".", "error") elif not name: flash("Enter the person's name.", "error") else: try: db().execute("UPDATE users SET role = ?, email = ?, name = ? WHERE id = ?", (role, email, name, uid)) db().commit() flash("User updated.", "ok") except sqlite3.IntegrityError: flash("That email is already used by another login.", "error") return redirect(url_for("users")) @app.post("/users//welcome") @roles_required("admin") def user_welcome(uid): target = db().execute("SELECT * FROM users WHERE id = ?", (uid,)).fetchone() if target is None or target["role"] == "admin" or not target["email"]: abort(404) try: send_welcome(target["name"], target["email"], target["role"]) flash(f"Welcome email sent to {target['email']}.", "ok") except MailError as e: flash(f"The welcome email was not sent: {e}.", "error") return redirect(url_for("users")) @app.post("/users//delete") @roles_required("admin") def user_delete(uid): target = db().execute("SELECT role FROM users WHERE id = ?", (uid,)).fetchone() if target is None: abort(404) if target["role"] == "admin": flash("Admin accounts cannot be deleted here.", "error") else: db().execute("DELETE FROM users WHERE id = ?", (uid,)) db().commit() flash("User deleted.", "ok") return redirect(url_for("users")) # ---- settings: welcome email template (admin) ---------------------------------------- def _email_context(): if g.user["role"] != "admin": return {} cfg = smtp_config() return {"subject": get_setting("email_subject", DEFAULT_SUBJECT), "body": get_setting("email_body", DEFAULT_BODY), "enabled": get_setting("email_enabled", "1") == "1", "mail_ready": cfg is not None, "otp_on": otp_enabled(), "mail_from": cfg["from_email"] if cfg else "", "template_vars": TEMPLATE_VARS} def _template_from_form(): subject = " ".join(request.form.get("subject", "").split())[:200] body = re.sub(r"(?is)", "", request.form.get("body", "")).strip() # no scripts in mail return subject, body[:50000] @app.post("/settings/signin") @roles_required("admin") def signin_save(): set_setting("otp_enabled", "1" if request.form.get("otp") == "on" else "0") db().commit() flash("Sign-in options saved.", "ok") return redirect(url_for("settings")) @app.post("/settings/email") @roles_required("admin") def email_save(): subject, body = _template_from_form() if not subject or not body: flash("The email needs both a subject and a body.", "error") else: set_setting("email_subject", subject) set_setting("email_body", body) set_setting("email_enabled", "1" if request.form.get("enabled") == "on" else "0") db().commit() flash("Welcome email template saved.", "ok") return redirect(url_for("settings")) @app.post("/settings/email/reset") @roles_required("admin") def email_reset(): db().execute("DELETE FROM settings WHERE key IN ('email_subject', 'email_body')") db().commit() flash("The welcome email template was reset to the default.", "ok") return redirect(url_for("settings")) @app.post("/settings/email/test") @roles_required("admin") def email_test(): """Send the template as currently typed (saved or not) to an address the admin chooses.""" to = request.form.get("to", "").strip().lower() subject, body = _template_from_form() if not FACULTY_EMAIL_RE.fullmatch(to): flash("Enter a valid email address to send the test to.", "error") elif not subject or not body: flash("The email needs both a subject and a body.", "error") else: values = {"app_name": APP_NAME, "name": "Sample Person", "email": to, "role": "Lead", "login_url": portal_login_url("lead"), "added_by": _me()[1]} try: send_mail(to, "[Test] " + fill_template(subject, values, False), fill_template(body, values, True)) flash(f"Test email sent to {to}.", "ok") except MailError as e: flash(f"The test email was not sent: {e}.", "error") return redirect(url_for("settings")) # ---- settings (change own password) -------------------------------------------------- @app.route("/settings", methods=["GET", "POST"]) @roles_required(*ANY) def settings(): if request.method == "POST" and g.user["role"] != "admin": flash("Your account signs in with Google, so there is no password to change.", "error") elif request.method == "POST": if not check_password_hash(g.user["password_hash"], request.form["current"]): flash("Current password is wrong.", "error") elif len(request.form["new"]) < 6: flash("New password must be 6+ characters.", "error") else: db().execute( "UPDATE users SET password_hash = ? WHERE id = ?", (generate_password_hash(request.form["new"]), g.user["id"]), ) db().commit() flash("Password changed.", "ok") return redirect(url_for("settings")) return render_template("settings.html", **_email_context()) if __name__ == "__main__": init_db() app.run(port=5001, debug=False)