"""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 mimetypes 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 html.parser import HTMLParser 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", "dot", "dotx", "rtf", "odt", "txt", "md", "log", "csv", "tsv", "xls", "xlsx", "xlsm", "ods", "ppt", "pptx", "pps", "ppsx", "odp", "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."), ("reminder", "Add Reminder", "calendar", "Track when availability was last updated and when it's due to be checked again."), ("faculty", "Add Faculty Specific KT", "user", "KT that belongs to one faculty member - pick them from the Faculty List."), ) 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", "logo_path"): 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") conn.execute( """ CREATE TABLE IF NOT EXISTS kt_criteria ( id INTEGER PRIMARY KEY AUTOINCREMENT, college_id INTEGER NOT NULL REFERENCES colleges(id) ON DELETE CASCADE, title TEXT NOT NULL, created_by INTEGER REFERENCES users(id) ON DELETE SET NULL, created_by_name TEXT ) """ ) conn.execute( """ CREATE TABLE IF NOT EXISTS issue_concerns ( id INTEGER PRIMARY KEY AUTOINCREMENT, college_id INTEGER NOT NULL REFERENCES colleges(id) ON DELETE CASCADE, title TEXT NOT NULL, description TEXT, image_path TEXT, created_at TEXT NOT NULL DEFAULT (datetime('now', 'localtime')), created_by INTEGER REFERENCES users(id) ON DELETE SET NULL, created_by_name TEXT ) """ ) if "pinned" not in [r[1] for r in conn.execute("PRAGMA table_info(kt_criteria)")]: conn.execute("ALTER TABLE kt_criteria ADD COLUMN pinned INTEGER NOT NULL DEFAULT 0") kt_cols = [r[1] for r in conn.execute("PRAGMA table_info(kts)")] if "criteria_id" not in kt_cols: conn.execute("ALTER TABLE kts ADD COLUMN criteria_id INTEGER REFERENCES kt_criteria(id) ON DELETE CASCADE") # every KT saved before this level existed WAS itself one criteria card with one block inside it: # give each of those a matching new kt_criteria row (using its old title), and rename the block # to its type's plain label so the title is not shown twice. block_label_fallback = {"description": "Description", "form": "Form", "attachment": "Attachment", "embed": "Embedded Media", "link": "Link", "alert": "Alert"} old_rows = conn.execute( "SELECT id, college_id, title, block_type, created_by, created_by_name FROM kts WHERE criteria_id IS NULL" ).fetchall() for old_id, college_id, old_title, block_type, created_by, created_by_name in old_rows: cur = conn.execute( "INSERT INTO kt_criteria (college_id, title, created_by, created_by_name) VALUES (?, ?, ?, ?)", (college_id, old_title or "Untitled", created_by, created_by_name), ) new_block_title = block_label_fallback.get(block_type, "Description") conn.execute("UPDATE kts SET criteria_id = ?, title = ? WHERE id = ?", (cur.lastrowid, new_block_title, old_id)) # every KT now belongs to a criteria, not a college directly - retire the old college_id # column (it also carried a NOT NULL constraint that ALTER TABLE can't lift on its own) conn.executescript( """ CREATE TABLE kts_new ( id INTEGER PRIMARY KEY AUTOINCREMENT, criteria_id INTEGER REFERENCES kt_criteria(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, created_by INTEGER REFERENCES users(id) ON DELETE SET NULL, created_by_name TEXT, block_type TEXT NOT NULL DEFAULT 'description', alert_level TEXT, file_path TEXT, file_name TEXT, file_type TEXT, file_size INTEGER ); INSERT INTO kts_new (id, criteria_id, title, text, kind, options, faculty_id, created_by, created_by_name, block_type, alert_level, file_path, file_name, file_type, file_size) SELECT id, criteria_id, title, text, kind, options, faculty_id, created_by, created_by_name, block_type, alert_level, file_path, file_name, file_type, file_size FROM kts; DROP TABLE kts; ALTER TABLE kts_new RENAME TO kts; """ ) kt_cols = [r[1] for r in conn.execute("PRAGMA table_info(kts)")] for col in ("description", "label"): if col not in kt_cols: conn.execute(f"ALTER TABLE kts ADD COLUMN {col} TEXT") kt_cols = [r[1] for r in conn.execute("PRAGMA table_info(kts)")] if "fields" not in kt_cols: conn.execute("ALTER TABLE kts ADD COLUMN fields TEXT") # a form block saved before multi-question support: its one label/kind/answer # becomes the first (and only) entry of its new fields list rows = conn.execute( "SELECT id, label, kind, text, options FROM kts WHERE block_type = 'form' AND label IS NOT NULL" ).fetchall() for row_id, row_label, row_kind, row_text, row_options in rows: entry = {"label": row_label, "kind": row_kind} if row_kind in ("multiple_choice", "checkboxes", "dropdown") and row_options: try: entry.update(json.loads(row_options)) except ValueError: entry["options"], entry["selected"] = [], [] else: entry["value"] = row_text or "" conn.execute("UPDATE kts SET fields = ? WHERE id = ?", (json.dumps([entry]), row_id)) 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)" _RICH_TAGS = {"b", "strong", "i", "em", "u", "ul", "ol", "li", "br", "div", "p", "span", "blockquote"} def _clean_rich_style(style): """Keep only text-align / margin-left declarations, with safe values - nothing else survives.""" if not style: return None kept = [] for decl in style.split(";"): prop, _, val = decl.partition(":") prop, val = prop.strip().lower(), val.strip().lower() if prop == "text-align" and val in ("left", "right", "center", "justify"): kept.append(f"text-align:{val}") elif prop == "margin-left": m = re.fullmatch(r"(\d{1,3})px", val) if m and int(m.group(1)) <= 400: kept.append(f"margin-left:{m.group(1)}px") elif prop == "margin": # shorthand (as Chrome's indent/outdent commands emit): 1-4 values, left is the last # when there are 4, otherwise the same as the horizontal value nums = re.findall(r"(\d{1,3})px|(0)(?!\w)", val) vals = [a or b for a, b in nums] if vals and re.fullmatch(r"(?:\d{1,3}px\s*|0\s*){1,4}", val): left = vals[-1] if len(vals) == 4 else (vals[1] if len(vals) >= 2 else vals[0]) if int(left) <= 400: kept.append(f"margin-left:{left}px") return "; ".join(kept) if kept else None class _RichTextSanitizer(HTMLParser): """Allowlist HTML sanitizer for the "Description" rich-text fields: only basic formatting tags survive (bold/italic/underline/lists/paragraphs/indent), and the only attribute kept anywhere is a style consisting solely of text-align and/or margin-left. Everything else - scripts, links, images, event handlers, any other tag or attribute - is stripped; disallowed tags lose their wrapper but keep their (escaped) text so nothing typed just disappears.""" def __init__(self): super().__init__(convert_charrefs=True) self.out = [] self.open_stack = [] def handle_starttag(self, tag, attrs): tag = tag.lower() if tag == "br": self.out.append("
") return if tag not in _RICH_TAGS: return style = _clean_rich_style(dict((k.lower(), v) for k, v in attrs if v is not None).get("style")) attr_str = f' style="{html.escape(style, quote=True)}"' if style else "" self.out.append(f"<{tag}{attr_str}>") self.open_stack.append(tag) def handle_startendtag(self, tag, attrs): if tag.lower() == "br": self.out.append("
") def handle_endtag(self, tag): tag = tag.lower() if tag in _RICH_TAGS and tag in self.open_stack: while self.open_stack[-1] != tag: self.out.append(f"") self.out.append(f"") self.open_stack.pop() def handle_data(self, data): self.out.append(html.escape(data)) def close(self): super().close() while self.open_stack: self.out.append(f"") def _sanitize_rich_text(raw, max_len=20000): """Reduce a pasted/typed rich-text editor body to a small safe HTML subset.""" parser = _RichTextSanitizer() try: parser.feed((raw or "")[: max_len * 2]) parser.close() except Exception: return html.escape(raw or "")[:max_len] return "".join(parser.out)[:max_len] def _rich_text_is_blank(sanitized_html): """True when sanitized rich-text HTML has no visible text left, only empty tags/whitespace.""" return not re.sub(r"<[^>]+>", "", sanitized_html or "").strip() 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) def _row_get(row, key): try: return row[key] except (KeyError, IndexError, TypeError): return None def _college_logo_url(c): """The logo to show for a college: its uploaded logo if it has one, else the automatic favicon.""" if _row_get(c, "logo_path"): return url_for("college_logo", cid=c["id"], v=c["logo_path"][5:17]) return _favicon_url(_row_get(c, "url")) app.jinja_env.globals["favicon_url"] = _favicon_url app.jinja_env.globals["logo_url"] = _college_logo_url LOGO_MAX_BYTES = 2 * 1024 * 1024 def _save_image(file_storage, prefix, max_bytes, noun): """Store an uploaded image. The type comes from the file's own bytes, never its name. Returns (stored_name, None) or (None, error).""" data = file_storage.read(max_bytes + 1) if not data: return None, f"That {noun} file is empty." if len(data) > max_bytes: return None, f"That {noun} is too big. The limit is {max_bytes // (1024 * 1024)} MB." if data.startswith(b"\x89PNG\r\n\x1a\n"): ext = "png" elif data.startswith(b"\xff\xd8\xff"): ext = "jpg" elif data[:6] in (b"GIF87a", b"GIF89a"): ext = "gif" elif data[:4] == b"RIFF" and data[8:12] == b"WEBP": ext = "webp" else: return None, f"The {noun} must be a PNG, JPG, GIF or WebP image." stored = prefix + secrets.token_hex(16) + "." + ext (UPLOAD_DIR / stored).write_bytes(data) return stored, None def _save_logo(file_storage): return _save_image(file_storage, "logo_", LOGO_MAX_BYTES, "logo") 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, attachments=(), reply_to=None): """attachments: (filename, bytes, mime_type) tuples.""" 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 if reply_to: msg["Reply-To"] = reply_to msg.set_content(html_to_text(body_html)) msg.add_alternative(body_html, subtype="html") for filename, data, mime in attachments: maintype, _, subtype = (mime or "application/octet-stream").partition("/") msg.add_attachment(data, maintype=maintype, subtype=subtype or "octet-stream", filename=filename) try: server = (smtplib.SMTP_SSL if cfg["port"] == 465 else smtplib.SMTP)( cfg["host"], cfg["port"], timeout=60 if attachments else 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(DISTINCT kc.college_id) FROM kt_criteria kc " "WHERE EXISTS (SELECT 1 FROM kts WHERE kts.criteria_id = kc.id)" ).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 kk JOIN kt_criteria kc2 ON kc2.id = kk.criteria_id " "WHERE kc2.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 _extract_iframe_src(raw): """If pasted text is a full