"""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, send_from_directory, session, url_for) from werkzeug.security import check_password_hash, generate_password_hash from werkzeug.utils import secure_filename 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") UPLOAD_DIR = Path(__file__).with_name("uploads") UPLOAD_DIR.mkdir(exist_ok=True) ALLOWED_UPLOAD_EXT = { "pdf", "doc", "docx", "ppt", "pptx", "xls", "xlsx", "txt", "csv", "png", "jpg", "jpeg", "gif", "webp", "zip", } MAX_UPLOAD_BYTES = 15 * 1024 * 1024 # 15 MB 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") BLOCK_TYPES = ( ("form", "Add Form", "list", "A single structured question: short answer, paragraph, multiple choice, checkboxes, drop-down, date or time."), ("attachment", "Add Attachment", "paperclip", "Upload a file for people to download - PDF, Word, Excel, PowerPoint, image, text, CSV or ZIP."), ("embed", "Add Embedded Media", "film", "Show a video or another page right on the card."), ("link", "Add Link", "link", "A button that opens another page in a new tab."), ("alert", "Add Alert", "alert", "A highlighted callout: Info, Warning or Critical."), ("description", "Add Description", "file-text", "Plain text - the simplest option."), ) BLOCK_META = {v: {"label": l, "icon": ic, "hint": h} for v, l, ic, h in BLOCK_TYPES} # US-only timezone list (the college portal covers US institutions), sourced from the IANA # tz database's official "US" entries (zone.tab), grouped by common name. TIMEZONE_GROUPS = ( ("Eastern Time", ( ("America/New_York", "New York (most areas)"), ("America/Detroit", "Detroit, Michigan"), ("America/Kentucky/Louisville", "Louisville, Kentucky"), ("America/Kentucky/Monticello", "Monticello, Kentucky (Wayne County)"), ("America/Indiana/Indianapolis", "Indianapolis, Indiana (most areas)"), ("America/Indiana/Vincennes", "Vincennes, Indiana"), ("America/Indiana/Winamac", "Winamac, Indiana (Pulaski County)"), ("America/Indiana/Marengo", "Marengo, Indiana (Crawford County)"), ("America/Indiana/Petersburg", "Petersburg, Indiana (Pike County)"), ("America/Indiana/Vevay", "Vevay, Indiana (Switzerland County)"), )), ("Central Time", ( ("America/Chicago", "Chicago (most areas)"), ("America/Indiana/Tell_City", "Tell City, Indiana (Perry County)"), ("America/Indiana/Knox", "Knox, Indiana (Starke County)"), ("America/Menominee", "Menominee, Michigan"), ("America/North_Dakota/Center", "Center, North Dakota (Oliver County)"), ("America/North_Dakota/New_Salem", "New Salem, North Dakota (Morton County)"), ("America/North_Dakota/Beulah", "Beulah, North Dakota (Mercer County)"), )), ("Mountain Time", ( ("America/Denver", "Denver (most areas)"), ("America/Boise", "Boise, Idaho / eastern Oregon"), ("America/Phoenix", "Phoenix, Arizona (no daylight saving)"), )), ("Pacific Time", ( ("America/Los_Angeles", "Los Angeles (most areas)"), )), ("Alaska Time", ( ("America/Anchorage", "Anchorage (most areas)"), ("America/Juneau", "Juneau"), ("America/Sitka", "Sitka"), ("America/Metlakatla", "Metlakatla (Annette Island)"), ("America/Yakutat", "Yakutat"), ("America/Nome", "Nome (western Alaska)"), ("America/Adak", "Adak (western Aleutians)"), )), ("Hawaii Time", ( ("Pacific/Honolulu", "Honolulu"), )), ) TIMEZONE_SET = frozenset(z for _, zones in TIMEZONE_GROUPS for z, _ in zones) 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__) def _icon(name, cls=""): """An inline SVG icon from the sprite in base.html, e.g. {{ icon('search') }}.""" from markupsafe import Markup return Markup(f'') app.jinja_env.globals["icon"] = _icon 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, url TEXT, location TEXT, timezone TEXT ); 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") college_cols = [r[1] for r in conn.execute("PRAGMA table_info(colleges)")] for col in ("url", "location", "timezone"): if col not in college_cols: conn.execute(f"ALTER TABLE colleges ADD COLUMN {col} 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") kt_cols = [r[1] for r in conn.execute("PRAGMA table_info(kts)")] if "block_type" not in kt_cols: conn.execute("ALTER TABLE kts ADD COLUMN block_type TEXT NOT NULL DEFAULT 'description'") # a KT saved before block types existed: one with a structured question type is now a "form" block conn.execute( "UPDATE kts SET block_type = 'form' WHERE kind IN " "('short', 'date', 'time', 'multiple_choice', 'checkboxes', 'dropdown')" ) for col in ("alert_level", "file_path", "file_name", "file_type"): if col not in kt_cols: conn.execute(f"ALTER TABLE kts ADD COLUMN {col} TEXT") if "file_size" not in kt_cols: conn.execute("ALTER TABLE kts ADD COLUMN file_size INTEGER") 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)" def _normalize_url(raw, empty_msg, invalid_msg, max_len=500): """Best-effort https:// URL. Returns (normalized_url_or_None, error_or_None).""" raw = (raw or "").strip() if not raw: return None, empty_msg if len(raw) > max_len: return None, "That web address is too long." candidate = raw if re.match(r"^https?://", raw, re.I) else "https://" + raw parsed = urllib.parse.urlparse(candidate) host = parsed.hostname or "" if parsed.scheme not in ("http", "https") or "." not in host or " " in candidate: return None, invalid_msg return candidate, None def _normalize_college_url(raw): return _normalize_url(raw, "Enter the college's website address.", "Enter a valid website address, e.g. https://www.college.edu", max_len=300) US_STATES = ( "Alabama", "Alaska", "Arizona", "Arkansas", "California", "Colorado", "Connecticut", "Delaware", "District of Columbia", "Florida", "Georgia", "Hawaii", "Idaho", "Illinois", "Indiana", "Iowa", "Kansas", "Kentucky", "Louisiana", "Maine", "Maryland", "Massachusetts", "Michigan", "Minnesota", "Mississippi", "Missouri", "Montana", "Nebraska", "Nevada", "New Hampshire", "New Jersey", "New Mexico", "New York", "North Carolina", "North Dakota", "Ohio", "Oklahoma", "Oregon", "Pennsylvania", "Rhode Island", "South Carolina", "South Dakota", "Tennessee", "Texas", "Utah", "Vermont", "Virginia", "Washington", "West Virginia", "Wisconsin", "Wyoming", ) US_STATE_SET = frozenset(US_STATES) def _split_location(location): """'City, State' -> (city, state). Best-effort split for redisplaying an existing college in the edit form; state is only recognised if it is an exact, known US state name.""" location = (location or "").strip() for state in US_STATES: suffix = ", " + state if location.endswith(suffix): return location[: -len(suffix)].strip(), state return location, "" def _favicon_url(college_url): """A small icon for the college, fetched by the browser from its own website - nothing to upload.""" if not college_url: return None host = urllib.parse.urlparse(college_url).hostname if not host: return None return "https://www.google.com/s2/favicons?sz=128&domain=" + urllib.parse.quote(host) app.jinja_env.globals["favicon_url"] = _favicon_url 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("dashboard")) 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("dashboard")) 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("dashboard")) 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("dashboard")) 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("dashboard")) @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"))) # ---- dashboard -------------------------------------------------------------- @app.route("/") def index(): return redirect(url_for("dashboard")) @app.route("/dashboard") @roles_required(*ANY) def dashboard(): d = db() stats = { "faculty": d.execute("SELECT count(*) FROM faculty").fetchone()[0], "colleges": d.execute("SELECT count(*) FROM colleges").fetchone()[0], "leads": d.execute("SELECT count(*) FROM users WHERE role = 'lead'").fetchone()[0], "users": d.execute("SELECT count(*) FROM users WHERE role = 'user'").fetchone()[0], "kts": d.execute("SELECT count(*) FROM kts").fetchone()[0], } loaded = d.execute("SELECT count(*) FROM colleges WHERE id IN (SELECT DISTINCT college_id FROM kts)").fetchone()[0] stats["colleges_with_kt"] = loaded stats["colleges_without_kt"] = stats["colleges"] - loaded rows = d.execute( "SELECT c.id, c.name, (SELECT count(*) FROM kts WHERE college_id = c.id) AS kt_count " "FROM colleges c ORDER BY kt_count DESC, c.name LIMIT 8" ).fetchall() return render_template("dashboard.html", stats=stats, rows=rows) # ---- search KT ------------------------------------------------------------ def _parse_desc(kind, payload): """Validate one 'Add Form' question. Returns (text, options_json, is_blank, error).""" if kind not in KIND_LABELS: return "", None, False, "choose a valid question 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 question content." 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 _human_size(n): if not n: return None size = float(n) for unit in ("B", "KB", "MB", "GB"): if size < 1024 or unit == "GB": return f"{size:.0f} {unit}" if unit == "B" else f"{size:.1f} {unit}" size /= 1024 return f"{size:.1f} GB" def _embed_src(url): """A YouTube/Vimeo watch link becomes its embeddable player URL; anything else embeds as typed.""" if not url: return None m = re.search(r"(?:youtube\.com/watch\?v=|youtu\.be/|youtube\.com/embed/)([\w-]{6,})", url) if m: return f"https://www.youtube.com/embed/{m.group(1)}" m = re.search(r"vimeo\.com/(\d+)", url) if m: return f"https://player.vimeo.com/video/{m.group(1)}" return url def _save_upload(file_storage): """Validate and store an uploaded attachment. Returns (info_dict, None) or (None, error).""" name = (file_storage.filename or "").strip() if not name: return None, "Choose a file to upload." ext = name.rsplit(".", 1)[-1].lower() if "." in name else "" if ext not in ALLOWED_UPLOAD_EXT: return None, "That file type is not allowed. Allowed: " + ", ".join(sorted(ALLOWED_UPLOAD_EXT)) + "." data = file_storage.read() if not data: return None, "That file is empty." if len(data) > MAX_UPLOAD_BYTES: return None, f"That file is too big. The limit is {MAX_UPLOAD_BYTES // (1024 * 1024)} MB." stored = secrets.token_hex(16) + "." + ext # never derived from the uploaded filename (UPLOAD_DIR / stored).write_bytes(data) safe_name = secure_filename(name) or ("file." + ext) return {"file_path": stored, "file_name": safe_name[:200], "file_size": len(data), "file_type": ext}, None def _delete_upload(stored_name): """Best-effort cleanup; never lets a missing/locked file break the caller.""" if not stored_name: return try: (UPLOAD_DIR / stored_name).unlink(missing_ok=True) except OSError: pass def _kt_view(row): """A KT row plus what the template needs to show it according to its block type.""" v = dict(row) v["block_type"] = v.get("block_type") or "description" 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 # Link: always a button. Embed: an inline player, with the same link as a fallback. # A plain description whose whole text is just a URL (saved before block types existed) # still gets the "Open link" button too, so nothing already stored changes how it looks. v["link"] = None if v["block_type"] in ("link", "embed"): v["link"] = v.get("text") elif v["block_type"] == "description" and v.get("kind") not in CHOICE_KINDS: candidate = (v.get("text") or "").strip() if re.fullmatch(r"https?://\S+", candidate): v["link"] = candidate v["embed_src"] = _embed_src(v.get("text")) if v["block_type"] == "embed" else None v["file_size_h"] = _human_size(v.get("file_size")) 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: return redirect(url_for("college_kt_page", cid=cid)) # a single college now has its own page 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) @app.route("/college//kt") @roles_required(*ANY) def college_kt_page(cid): """A standalone, printable-looking page of one college's saved KT (opened in its own tab).""" college = db().execute("SELECT * FROM colleges WHERE id = ?", (cid,)).fetchone() if college is None: abort(404) kts = [_kt_view(k) 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 " "FROM kts LEFT JOIN faculty f ON f.id = kts.faculty_id LEFT JOIN users ku ON ku.id = kts.created_by " "WHERE kts.college_id = ? ORDER BY kts.id", (cid,) )] return render_template("college_kt_page.html", college=college, kts=kts, standalone=True) # ---- 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() def _college_form(): """Read and validate the college form. Returns (values, error_or_None); values redisplay the form on error.""" f = { "name": request.form.get("name", "").strip(), "url_raw": request.form.get("url", "").strip(), "city": request.form.get("city", "").strip()[:100], "state": request.form.get("state", "").strip(), "timezone": request.form.get("timezone", "").strip(), } url, url_error = _normalize_college_url(f["url_raw"]) f["url"] = url or f["url_raw"] # keep exactly what they typed if it did not validate f["location"] = f"{f['city']}, {f['state']}" if f["city"] and f["state"] else (f["city"] or f["state"]) error = None if not f["name"]: error = "College name is required." elif url_error: error = url_error elif not f["city"]: error = "Enter the city." elif f["state"] not in US_STATE_SET: error = "Choose a state from the list." elif f["timezone"] not in TIMEZONE_SET: error = "Choose a timezone from the list." return f, error @app.route("/colleges/new", methods=["GET", "POST"]) @roles_required(*EDITORS) def college_new(): values = {} if request.method == "POST": values, error = _college_form() if error: flash(error, "error") else: try: db().execute( "INSERT INTO colleges (name, url, location, timezone, created_by, created_by_name) " "VALUES (?, ?, ?, ?, ?, ?)", (values["name"], values["url"], values["location"], values["timezone"], *_me()), ) db().commit() flash(f"College {values['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(), f=values, timezone_groups=TIMEZONE_GROUPS, us_states=US_STATES) @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) city, state = _split_location(row["location"]) values = {"name": row["name"], "url": row["url"] or "", "url_raw": row["url"] or "", "city": city, "state": state, "timezone": row["timezone"] or ""} if request.method == "POST": values, error = _college_form() if error: flash(error, "error") else: try: db().execute( "UPDATE colleges SET name = ?, url = ?, location = ?, timezone = ? WHERE id = ?", (values["name"], values["url"], values["location"], values["timezone"], cid), ) db().commit() flash("College updated.", "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=values, cid=cid, timezone_groups=TIMEZONE_GROUPS, us_states=US_STATES) @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 -------------------------------------------------------- # Clicking "Add KT Criteria" first shows a picker of six block types (Form, Attachment, # Embedded Media, Link, Alert, Description). Picking one loads a small form for just that # type. Editing a KT shows the same fields its own block type already uses - the type itself # cannot be changed after saving, only its content. def _kt_form_values(block_type, existing=None): """What the form fields should show: posted values on a failed submission, else the existing KT (editing), else blanks (a fresh add).""" if request.method == "POST": try: payload = json.loads(request.form.get("payload", "{}")) if not isinstance(payload, dict): payload = {} except ValueError: payload = {} return { "title": request.form.get("title", "").strip(), "text": request.form.get("kt", "").strip(), "url": request.form.get("url", "").strip(), "kind": request.form.get("kind") or "paragraph", "payload": payload, "level": request.form.get("level", ""), } if existing: payload = {"value": existing["text"]} if existing["kind"] in CHOICE_KINDS and existing["options"]: try: payload = json.loads(existing["options"]) except ValueError: pass return { "title": existing["title"], "text": existing["text"], "url": existing["text"], "kind": existing["kind"] or "paragraph", "payload": payload, "level": existing["alert_level"] or "", } return {"title": "", "text": "", "url": "", "kind": "paragraph", "payload": {"value": ""}, "level": ""} def _parse_kt(block_type, existing=None): """Read and validate the Add/Edit KT form for one block type. Returns (fields_ready_for_the_database, error_or_None).""" title = request.form.get("title", "").strip()[:300] out = { "title": title, "text": "", "kind": None, "options": None, "alert_level": None, "file_path": existing["file_path"] if existing else None, "file_name": existing["file_name"] if existing else None, "file_size": existing["file_size"] if existing else None, "file_type": existing["file_type"] if existing else None, } if not title: return None, "Enter a title." if block_type == "description": text = request.form.get("kt", "").strip()[:10000] if not text: return None, "Enter the description." out.update(text=text, kind="paragraph") elif block_type == "form": kind = request.form.get("kind", "") try: payload = json.loads(request.form.get("payload", "{}")) if not isinstance(payload, dict): payload = {} except ValueError: payload = {} text, options, _blank, err = _parse_desc(kind, payload) if err: return None, "Question: " + err out.update(text=text, kind=kind, options=options) elif block_type in ("link", "embed"): url, err = _normalize_url(request.form.get("url", ""), "Enter a web address.", "Enter a valid web address, e.g. https://example.com") if err: return None, err out.update(text=url, kind=block_type) elif block_type == "alert": level = request.form.get("level", "") message = request.form.get("kt", "").strip()[:2000] if level not in ("info", "warning", "critical"): return None, "Choose an alert level." if not message: return None, "Enter the alert message." out.update(text=message, alert_level=level, kind="alert") elif block_type == "attachment": out["kind"] = "attachment" f = request.files.get("file") if f is not None and f.filename: info, err = _save_upload(f) if err: return None, err if existing and existing["file_path"]: _delete_upload(existing["file_path"]) # replacing a file: drop the old one out.update(file_path=info["file_path"], file_name=info["file_name"], file_size=info["file_size"], file_type=info["file_type"]) elif not (existing and existing["file_path"]): return None, "Choose a file to upload." # else: editing without picking a new file - keep the one already stored else: return None, "Choose what kind of KT criteria to add." return out, None @app.route("/kt/new", methods=["GET", "POST"]) @roles_required(*EDITORS) def kt_new(): selected = request.values.get("college", type=int) block_type = (request.form.get("block_type") if request.method == "POST" else request.args.get("type", "")).strip() if block_type not in BLOCK_META: block_type = None selected_college = db().execute("SELECT * FROM colleges WHERE id = ?", (selected,)).fetchone() if selected else None values = _kt_form_values(block_type) if request.method == "POST" and block_type: fields, error = _parse_kt(block_type) if not selected_college: error = error or "Please choose a college." if error: flash(error, "error") else: db().execute( "INSERT INTO kts (college_id, title, text, kind, options, alert_level, file_path, file_name, " "file_size, file_type, block_type, created_by, created_by_name) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", (selected, fields["title"], fields["text"], fields["kind"], fields["options"], fields["alert_level"], fields["file_path"], fields["file_name"], fields["file_size"], fields["file_type"], block_type, *_me()), ) db().commit() flash("KT saved.", "ok") return redirect(url_for("search", q=selected_college["name"])) return render_template("kt_form.html", colleges=_college_rows(), selected=selected, selected_college=selected_college, kt=None, values=values, block_type=block_type, block_types=BLOCK_TYPES, block_meta=BLOCK_META, 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) block_type = kt["block_type"] or "description" values = _kt_form_values(block_type, existing=kt) if request.method == "POST": fields, error = _parse_kt(block_type, existing=kt) if error: flash(error, "error") else: db().execute( "UPDATE kts SET title = ?, text = ?, kind = ?, options = ?, alert_level = ?, " "file_path = ?, file_name = ?, file_size = ?, file_type = ?, faculty_id = NULL WHERE id = ?", (fields["title"], fields["text"], fields["kind"], fields["options"], fields["alert_level"], fields["file_path"], fields["file_name"], fields["file_size"], fields["file_type"], 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"], selected_college=None, kt=kt, values=values, block_type=block_type, block_types=BLOCK_TYPES, block_meta=BLOCK_META, kinds=KINDS) @app.post("/kt//delete") @roles_required(*EDITORS) def kt_delete(kid): kt = db().execute("SELECT file_path FROM kts WHERE id = ?", (kid,)).fetchone() db().execute("DELETE FROM kts WHERE id = ?", (kid,)) db().commit() if kt and kt["file_path"]: _delete_upload(kt["file_path"]) flash("KT deleted.", "ok") return redirect(url_for("search")) @app.route("/kt//file") @roles_required(*ANY) def kt_file(kid): kt = db().execute("SELECT * FROM kts WHERE id = ?", (kid,)).fetchone() if kt is None or kt["block_type"] != "attachment" or not kt["file_path"]: abort(404) return send_from_directory(UPLOAD_DIR, kt["file_path"], download_name=kt["file_name"] or kt["file_path"]) # ---- 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)