"""Accounts (email + password, PBKDF2, SQLite) and admin helpers.""" import hashlib, hmac, html, os, re, secrets, shutil, sqlite3, tempfile, threading, time from pathlib import Path import gradio as gr OUT_DIR = Path(tempfile.gettempdir()) / "cs_outputs" OUT_DIR.mkdir(parents=True, exist_ok=True) # --------------------------------------------------------------------------- # Accounts (email + password, PBKDF2-hashed, SQLite) # Stored in /data when persistent storage is enabled; otherwise next to app.py. # --------------------------------------------------------------------------- _PERSISTENT = Path("/data").is_dir() and os.access("/data", os.W_OK) _DATA_DIR = Path("/data") if _PERSISTENT else Path(__file__).parent # The Space's /data is a mounted storage bucket. SQLite must not be opened there: its locking and in-place page # writes are not safe on such a mount and the file ends up "malformed". So the live database is on the container's # own disk (USERS_DB) and a verified snapshot of it is kept in /data (STORE_DB): read at start, rewritten after # every change. USERS_DB set by hand (tests, another host) is used as it is, with no snapshot. STORE_DB = None if os.getenv("USERS_DB"): USERS_DB = Path(os.environ["USERS_DB"]) elif _PERSISTENT: STORE_DB = _DATA_DIR / "users.db" USERS_DB = Path(os.getenv("USERS_LIVE_DIR", str(Path(tempfile.gettempdir()) / "cs_live"))) / "users.db" else: USERS_DB = _DATA_DIR / "users.db" _EMAIL_RE = re.compile(r"^[^@\s]+@[^@\s]+\.[^@\s]+$") # Optional backup of the user database to a PRIVATE Hugging Face dataset repo, so accounts # survive restarts/rebuilds even without Persistent storage. # USERS_REPO e.g. "Rasheed/suite-users" (created automatically, private) # USERS_REPO_TOKEN write token (falls back to HF_TOKEN) USERS_REPO = os.getenv("USERS_REPO", "").strip() _REPO_TOKEN = os.getenv("USERS_REPO_TOKEN") or os.getenv("HF_TOKEN") STORAGE_NOTE = "" _sync_now = threading.Event() def _restore_backup(): global STORAGE_NOTE if not USERS_REPO: STORAGE_NOTE = "" if _PERSISTENT else ( "⚠️ Accounts are stored on temporary disk and are erased when the Space restarts. " "Enable Persistent storage or set the USERS_REPO secret.") return try: from huggingface_hub import HfApi, hf_hub_download HfApi(token=_REPO_TOKEN).create_repo(USERS_REPO, repo_type="dataset", private=True, exist_ok=True) if not USERS_DB.exists(): p = hf_hub_download(USERS_REPO, "users.db", repo_type="dataset", token=_REPO_TOKEN) shutil.copy(p, USERS_DB) print(f"[auth] restored users.db from {USERS_REPO}") except Exception as err: # first run: file not in repo yet print(f"[auth] backup restore skipped: {err}") def _backup_loop(): from huggingface_hub import HfApi api, last = HfApi(token=_REPO_TOKEN), None while True: _sync_now.wait(timeout=60) _sync_now.clear() try: if not USERS_DB.exists(): continue m = USERS_DB.stat().st_mtime if m == last: continue snap = Path(tempfile.gettempdir()) / "users_backup.db" src, dst = sqlite3.connect(str(USERS_DB)), sqlite3.connect(str(snap)) with dst: src.backup(dst) # consistent copy even during writes src.close(); dst.close() api.upload_file(path_or_fileobj=str(snap), path_in_repo="users.db", repo_id=USERS_REPO, repo_type="dataset", commit_message="users backup") last = m except Exception as err: print(f"[auth] backup failed: {err}") time.sleep(30) def _healthy(path): """True when SQLite reads the whole file without finding damage.""" try: con = sqlite3.connect(str(path), timeout=10) try: return con.execute("PRAGMA integrity_check").fetchone()[0] == "ok" finally: con.close() except sqlite3.Error: return False def _salvage(bad, good): """Copy every row that can still be read from a damaged database into a new one. Returns {table: rows}.""" src = sqlite3.connect(f"file:{Path(bad).as_posix()}?mode=ro", uri=True, timeout=10) dst = sqlite3.connect(str(good)) saved = {} try: tables = src.execute("SELECT name, sql FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%'" ).fetchall() for name, sql in tables: try: dst.execute(sql) keyed = any(r[5] for r in src.execute(f'PRAGMA table_info("{name}")')) except sqlite3.Error as err: print(f"[auth] salvage: table {name} skipped: {err}") continue n = 0 # from the start, then (when rows have a key, so none is copied twice) from the end: what lies on # either side of a damaged page is kept. NOT INDEXED: a damaged index must not hide good rows. for order in ("ASC", "DESC") if keyed else ("ASC",): try: for row in src.execute(f'SELECT * FROM "{name}" NOT INDEXED ORDER BY rowid {order}'): cur = dst.execute(f'INSERT OR IGNORE INTO "{name}" VALUES ({",".join("?" * len(row))})', row) n += cur.rowcount break except sqlite3.Error as err: print(f"[auth] salvage: {name} read stopped ({order}): {err}") saved[name] = n dst.commit() finally: src.close() dst.close() return saved def _load_store(): """Start of the app: bring the snapshot in /data to the live database, repairing it when it is damaged.""" global STORAGE_NOTE if not STORE_DB: return USERS_DB.parent.mkdir(parents=True, exist_ok=True) if USERS_DB.exists() or not STORE_DB.exists(): return tmp = USERS_DB.with_name("incoming.db") try: for suffix in ("", "-journal", "-wal"): # an unfinished write of the old in-place database p = Path(str(STORE_DB) + suffix) if p.exists(): shutil.copy(p, str(tmp) + suffix) if _healthy(tmp): for suffix in ("-journal", "-wal", "-shm"): Path(str(tmp) + suffix).unlink(missing_ok=True) os.replace(tmp, USERS_DB) print(f"[auth] accounts loaded from {STORE_DB}") return keep = STORE_DB.with_name(f"users.damaged-{time.strftime('%Y%m%d-%H%M%S')}.db") shutil.copy(STORE_DB, keep) # the damaged file is kept, never thrown away saved = _salvage(tmp, USERS_DB) print(f"[auth] {STORE_DB} was damaged; rows recovered: {saved}; the damaged file is kept as {keep}") except Exception as err: print(f"[auth] could not load {STORE_DB}: {err}") STORAGE_NOTE = "⚠️ The saved accounts could not be read at start. Please contact support before signing up again." finally: for suffix in ("", "-journal", "-wal", "-shm"): Path(str(tmp) + suffix).unlink(missing_ok=True) _store_lock, _stored = threading.Lock(), [None] def store_now(): """Write a consistent, checked snapshot of the live database to /data (whole file, then renamed into place).""" if not STORE_DB or not USERS_DB.exists(): return False with _store_lock: st = USERS_DB.stat() stamp = (st.st_mtime_ns, st.st_size) if stamp == _stored[0]: return False snap = USERS_DB.with_name("snapshot.db") src, dst = sqlite3.connect(str(USERS_DB), timeout=10), sqlite3.connect(str(snap)) try: with dst: src.backup(dst) # consistent copy even during writes finally: src.close() dst.close() if not _healthy(snap): print("[auth] snapshot failed its check; the stored copy is left as it was") return False new = STORE_DB.with_name("users.db.new") try: shutil.copyfile(snap, new) os.replace(new, STORE_DB) except OSError: # a mount that cannot rename: write the file directly shutil.copyfile(snap, STORE_DB) for suffix in ("-journal", "-wal", "-shm"): # left by the old in-place database Path(str(STORE_DB) + suffix).unlink(missing_ok=True) _stored[0] = stamp return True def _store_loop(): while True: _store_now.wait(timeout=15) _store_now.clear() try: store_now() except Exception as err: print(f"[auth] storing the accounts failed: {err}") time.sleep(15) _store_now = threading.Event() def request_sync(): _sync_now.set() _store_now.set() _load_store() _restore_backup() if STORE_DB: import atexit threading.Thread(target=_store_loop, daemon=True).start() atexit.register(lambda: store_now()) if USERS_REPO: threading.Thread(target=_backup_loop, daemon=True).start() def _db(): con = sqlite3.connect(str(USERS_DB), timeout=10) con.execute( "CREATE TABLE IF NOT EXISTS users (email TEXT PRIMARY KEY, name TEXT, salt TEXT, " "pw TEXT, created TEXT DEFAULT CURRENT_TIMESTAMP, last_login TEXT, runs INTEGER DEFAULT 0)" ) con.execute("CREATE TABLE IF NOT EXISTS meta (k TEXT PRIMARY KEY, v TEXT)") con.execute("CREATE TABLE IF NOT EXISTS usage (who TEXT, day TEXT, kind TEXT, n INTEGER DEFAULT 0, " "PRIMARY KEY (who, day, kind))") cols = {r[1] for r in con.execute("PRAGMA table_info(users)")} if "active" not in cols: con.execute("ALTER TABLE users ADD COLUMN active INTEGER DEFAULT 1") if "text_runs" not in cols: con.execute("ALTER TABLE users ADD COLUMN text_runs INTEGER DEFAULT 0") for c in ("doc_limit", "text_limit"): # per-user overrides (NULL = default) if c not in cols: con.execute(f"ALTER TABLE users ADD COLUMN {c} INTEGER") return con # --------------------------------------------------------------------------- # "Stay signed in" tokens (kept in the visitor's browser, HMAC-signed, 30 days) # --------------------------------------------------------------------------- SESSION_DAYS = int(os.getenv("SESSION_DAYS", "30")) def session_secret(): env = os.getenv("SESSION_SECRET") if env: return env with _db() as con: row = con.execute("SELECT v FROM meta WHERE k='secret'").fetchone() if row: return row[0] s = secrets.token_hex(32) con.execute("INSERT INTO meta (k, v) VALUES ('secret', ?)", (s,)) request_sync() return s def make_token(user): exp = int(time.time()) + SESSION_DAYS * 86400 body = f"{user['email']}|{exp}" sig = hmac.new(session_secret().encode(), body.encode(), hashlib.sha256).hexdigest() return f"{body}|{sig}" def user_from_token(token): try: email, exp, sig = (token or "").rsplit("|", 2) body = f"{email}|{exp}" good = hmac.new(session_secret().encode(), body.encode(), hashlib.sha256).hexdigest() if not hmac.compare_digest(sig, good) or int(exp) < time.time(): return None with _db() as con: row = con.execute("SELECT name, active FROM users WHERE email=?", (email,)).fetchone() if not row or row[1] == 0: return None return {"email": email, "name": row[0]} except Exception: return None # Admins: comma-separated emails in the ADMIN_EMAILS secret (Space settings → Variables and secrets). ADMIN_EMAILS = {e.strip().lower() for e in os.getenv("ADMIN_EMAILS", "").split(",") if e.strip()} def is_admin(user): return bool(user and user.get("email", "").lower() in ADMIN_EMAILS) def _hash(password, salt): return hashlib.pbkdf2_hmac("sha256", password.encode(), bytes.fromhex(salt), 200_000).hex() def create_user(name, email, password, confirm): name, email = (name or "").strip(), (email or "").strip().lower() if not name: return None, "Please enter your name." if not _EMAIL_RE.match(email): return None, "Please enter a valid email address." if len(password or "") < 8: return None, "Password must be at least 8 characters." if password != confirm: return None, "Passwords do not match." salt = secrets.token_hex(16) with _db() as con: if con.execute("SELECT 1 FROM users WHERE email=?", (email,)).fetchone(): return None, "An account with this email already exists. Please log in." con.execute("INSERT INTO users (email, name, salt, pw, last_login) VALUES (?,?,?,?,CURRENT_TIMESTAMP)", (email, name, salt, _hash(password, salt))) request_sync() return {"email": email, "name": name}, None # Wrong passwords per account: after LOGIN_TRIES within LOGIN_WINDOW seconds the account's log-in waits out the # window, so a password cannot be guessed by trying thousands of them. (Kept in memory: a restart clears it.) LOGIN_TRIES = int(os.getenv("LOGIN_TRIES", "8")) LOGIN_WINDOW = int(os.getenv("LOGIN_WINDOW", "900")) _FAILS, _fails_lock = {}, threading.Lock() def _recent_fails(email, add=False, clear=False): now = time.time() with _fails_lock: if clear: _FAILS.pop(email, None) return 0 times = [t for t in _FAILS.get(email, []) if now - t < LOGIN_WINDOW] if add: times.append(now) if times: _FAILS[email] = times else: _FAILS.pop(email, None) if len(_FAILS) > 20000: # never grows without bound _FAILS.clear() return len(times) def check_user(email, password): email = (email or "").strip().lower() if _recent_fails(email) >= LOGIN_TRIES: return None, (f"Too many wrong passwords for this account. Please wait {LOGIN_WINDOW // 60} minutes, " "or use 🔑 Forgot password.") with _db() as con: row = con.execute("SELECT name, salt, pw, active FROM users WHERE email=?", (email,)).fetchone() if not row: return None, "No account found with this email. Please check it or create an account under Sign up." if not hmac.compare_digest(row[2], _hash(password or "", row[1])): _recent_fails(email, add=True) return None, "Incorrect password. Please try again, or use 🔑 Forgot password." if row[3] == 0: return None, "This account has been disabled. Please contact CS Software Solutions." con.execute("UPDATE users SET last_login=CURRENT_TIMESTAMP WHERE email=?", (email,)) _recent_fails(email, clear=True) return {"email": email, "name": row[0]}, None def _count_run(user, document=True): """Increment the user's document or pasted-text run counter (no-op for guests).""" if not user or not user.get("email"): return col = "runs" if document else "text_runs" try: with _db() as con: con.execute(f"UPDATE users SET {col} = COALESCE({col}, 0) + 1 WHERE email=?", (user["email"],)) except Exception: pass class Upsell(gr.Error): """Not an error: a plan / sign-in / limit message. The app shows it as a friendly centred pop-up. kind: "upgrade" (paid plan unlocks it), "signin" (free account unlocks it), "limit" (try again later).""" def __init__(self, message, kind="upgrade", title=None): titles = {"upgrade": "Unlock more with a plan", "signin": "Create your free account", "limit": "You've reached today's limit"} super().__init__(message, title=title or titles.get(kind, "Heads up"), print_exception=False) self.kind, self.headline, self.text = kind, title or titles.get(kind, "Heads up"), message def _require(user): if not user or not user.get("email"): raise Upsell("Sign in or create a free account to upload Word and PDF documents — it takes 20 seconds. " "Pasting text works without an account.", kind="signin", title="Upload documents for free") with _db() as con: row = con.execute("SELECT active FROM users WHERE email=?", (user["email"],)).fetchone() if not row or row[0] == 0: raise gr.Error("This account has been disabled.") def _msg(text, ok=False): cls = "cs-ok" if ok else "cs-bad" return f'
{text}
' if text else "" # --------------------------------------------------------------------------- # Admin panel # --------------------------------------------------------------------------- ADMIN_COLS = ["Name", "Email", "Joined", "Last login", "Doc runs", "Text runs", "Status", "Role"] def _need_admin(user): if not is_admin(user): raise gr.Error("Admin access only.") def admin_stats(user): _need_admin(user) with _db() as con: total = con.execute("SELECT COUNT(*) FROM users").fetchone()[0] active = con.execute("SELECT COUNT(*) FROM users WHERE active=1").fetchone()[0] runs = con.execute("SELECT COALESCE(SUM(runs),0) + COALESCE(SUM(text_runs),0) FROM users").fetchone()[0] new7 = con.execute("SELECT COUNT(*) FROM users WHERE created >= datetime('now','-7 days')").fetchone()[0] seen7 = con.execute("SELECT COUNT(*) FROM users WHERE last_login >= datetime('now','-7 days')").fetchone()[0] tiles = [("Total users", total), ("Active accounts", active), ("Total runs", runs), ("New (7 days)", new7), ("Logged in (7 days)", seen7)] cells = "".join(f'
{v:,}
{k}
' for k, v in tiles) return f'
{cells}
' def admin_table(user, query=""): _need_admin(user) q = f"%{(query or '').strip().lower()}%" with _db() as con: rows = con.execute( "SELECT name, email, created, last_login, runs, COALESCE(text_runs,0), active FROM users " "WHERE lower(name) LIKE ? OR lower(email) LIKE ? ORDER BY created DESC", (q, q)).fetchall() return [[n, e, (c or "")[:16], (l or "")[:16], r, t, "Active" if a else "Disabled", "Admin" if e in ADMIN_EMAILS else "User"] for n, e, c, l, r, t, a in rows] def _admin_refresh(user, query, note, ok=True): request_sync() return admin_stats(user), admin_table(user, query), _msg(note, ok=ok) def admin_search(user, query): _need_admin(user) return admin_table(user, query) def admin_set_active(user, email, query, active): _need_admin(user) email = (email or "").strip().lower() if not email: return _admin_refresh(user, query, "Enter the user's email first.", ok=False) if email in ADMIN_EMAILS and not active: return _admin_refresh(user, query, "Admin accounts can't be disabled here.", ok=False) with _db() as con: n = con.execute("UPDATE users SET active=? WHERE email=?", (1 if active else 0, email)).rowcount if not n: return _admin_refresh(user, query, f"No user found with {email}.", ok=False) return _admin_refresh(user, query, f"{email} is now {'active' if active else 'disabled'}.") def admin_reset_pw(user, email, new_pw, query): _need_admin(user) email = (email or "").strip().lower() if len(new_pw or "") < 8: return _admin_refresh(user, query, "New password must be at least 8 characters.", ok=False) salt = secrets.token_hex(16) with _db() as con: n = con.execute("UPDATE users SET salt=?, pw=? WHERE email=?", (salt, _hash(new_pw, salt), email)).rowcount if n: with _pw_db() as con: con.execute("UPDATE reset_requests SET status='done' WHERE email=? AND status='open'", (email,)) request_sync() if not n: return _admin_refresh(user, query, f"No user found with {email}.", ok=False) return _admin_refresh(user, query, f"Password reset for {email}. Share it with them securely.") def admin_delete(user, email, confirm, query): _need_admin(user) email = (email or "").strip().lower() if (confirm or "").strip().lower() != email or not email: return _admin_refresh(user, query, "Type the same email in the confirm box to delete.", ok=False) if email in ADMIN_EMAILS: return _admin_refresh(user, query, "Admin accounts can't be deleted here.", ok=False) with _db() as con: n = con.execute("DELETE FROM users WHERE email=?", (email,)).rowcount if not n: return _admin_refresh(user, query, f"No user found with {email}.", ok=False) return _admin_refresh(user, query, f"Deleted {email}.") def admin_export(user): _need_admin(user) import csv folder = Path(tempfile.mkdtemp(dir=OUT_DIR)) path = folder / "users.csv" with open(path, "w", newline="", encoding="utf-8") as f: w = csv.writer(f) w.writerow(ADMIN_COLS) w.writerows(admin_table(user)) return str(path) # --------------------------------------------------------------------------- # Daily usage limits (0 = unlimited). Admins are never limited. # DAILY_DOC_LIMIT document runs per signed-in user per day (default 20) # DAILY_TEXT_LIMIT pasted-text runs per signed-in user per day (default 100) # DAILY_GUEST_TEXT_LIMIT pasted-text runs per guest (by IP) per day (default 15) # --------------------------------------------------------------------------- LIMITS = { ("user", "doc"): int(os.getenv("DAILY_DOC_LIMIT", "20")), ("user", "text"): int(os.getenv("DAILY_TEXT_LIMIT", "100")), ("guest", "text"): int(os.getenv("DAILY_GUEST_TEXT_LIMIT", "15")), } def _who(user, request): if user and user.get("email"): return user["email"] ip = "" try: h = request.headers if request is not None else {} ip = (h.get("x-forwarded-for") or "").split(",")[0].strip() or getattr(request.client, "host", "") except Exception: pass return "ip:" + (ip or "unknown") DAY_OFFSET_MIN = int(os.getenv("DAY_TZ_OFFSET_MIN", "330")) # daily limits follow the Indian day (IST) def _today(): return time.strftime("%Y-%m-%d", time.gmtime(time.time() + DAY_OFFSET_MIN * 60)) def usage_today(who, kind): with _db() as con: row = con.execute("SELECT n FROM usage WHERE who=? AND day=? AND kind=?", (who, _today(), kind)).fetchone() return row[0] if row else 0 def user_limits(email): """Effective daily limits for a signed-in user: {'doc': n, 'text': n} (0 = unlimited).""" lim = {"doc": LIMITS[("user", "doc")], "text": LIMITS[("user", "text")]} try: with _db() as con: row = con.execute("SELECT doc_limit, text_limit FROM users WHERE email=?", (email,)).fetchone() if row: lim["doc"] = lim["doc"] if row[0] is None else row[0] lim["text"] = lim["text"] if row[1] is None else row[1] except Exception: pass return lim def admin_set_limits(user, email, doc, text, query=""): """Blank = back to default, 0 = unlimited.""" _need_admin(user) email = (email or "").strip().lower() conv = lambda v: None if v in (None, "") else int(float(v)) try: d, t = conv(doc), conv(text) except ValueError: return _admin_refresh(user, query, "Limits must be whole numbers (blank = default, 0 = unlimited).", ok=False) with _db() as con: n = con.execute("UPDATE users SET doc_limit=?, text_limit=? WHERE email=?", (d, t, email)).rowcount if not n: return _admin_refresh(user, query, f"No user found with {email}.", ok=False) show = lambda v: "default" if v is None else ("unlimited" if v == 0 else str(v)) return _admin_refresh(user, query, f"Limits for {email}: documents {show(d)}, texts {show(t)} per day.") def _override(email, col): try: with _db() as con: row = con.execute(f"SELECT {col} FROM users WHERE email=?", (email,)).fetchone() return row[0] if row else None except Exception: return None def check_limit(user, request, document): """Raise a friendly error if today's limit is used up.""" if is_admin(user): return import plans as P kind = "doc" if document else "text" role = "user" if user else "guest" if kind == "text": override = _override(user["email"], "text_limit") if user else None lim = override if override is not None else P.daily_paste_limit(user) else: # monthly quota lives in plans.py; this is only an anti-abuse cap override = _override(user["email"], "doc_limit") if user else None st = P.state(user) if user else None if override is None and st and st["key"] in P.PAID: return # paid plans: the daily document allowance is checked in plans.check_doc lim = override if override is not None else LIMITS[("user", "doc")] if lim <= 0: return if usage_today(_who(user, request), kind) >= lim: what = "document runs" if document else "pasted-text runs" if not user: raise Upsell(f"Guests can do {lim} {what} a day. Create a free account for a higher limit — " "or come back after midnight (IST).", kind="signin", title="You've reached today's free limit") paid = P.state(user)["key"] in P.PAID raise Upsell(f"You've used all {lim} {what} for today. It resets at midnight (IST)." + ("" if paid else " Paid plans have much higher daily limits."), kind="limit" if paid else "upgrade", title="You've reached today's limit") def add_usage(user, request, document): kind = "doc" if document else "text" try: with _db() as con: con.execute("INSERT INTO usage (who, day, kind, n) VALUES (?,?,?,1) " "ON CONFLICT(who, day, kind) DO UPDATE SET n = n + 1", (_who(user, request), _today(), kind)) con.execute("DELETE FROM usage WHERE day < date('now','-7 days')") except Exception: pass # --------------------------------------------------------------------------- # Passwords without email: # * Recovery key — shown once at sign-up (and on demand in Account); stored only as a hash. # Forgot password = email + recovery key + new password. A used key is replaced by a new one. # * Change password — signed-in users (current password + new password). # * No key? The user sends a reset request; the admin sees it in Admin -> Users and sets a # temporary password, shared by WhatsApp/phone. # --------------------------------------------------------------------------- KEY_ALPHABET = "ABCDEFGHJKLMNPQRSTUVWXYZ23456789" # no 0/O/1/I confusion MAX_KEY_FAILS = 5 # wrong keys per email per hour def _pw_db(): con = _db() cols = {r[1] for r in con.execute("PRAGMA table_info(users)")} if "recovery_hash" not in cols: con.execute("ALTER TABLE users ADD COLUMN recovery_hash TEXT") con.execute("CREATE TABLE IF NOT EXISTS reset_fails (email TEXT, ts REAL)") con.execute("CREATE TABLE IF NOT EXISTS reset_requests (id INTEGER PRIMARY KEY AUTOINCREMENT, " "ts TEXT DEFAULT CURRENT_TIMESTAMP, email TEXT, contact TEXT, status TEXT DEFAULT 'open')") return con def _key_hash(email, key): k = re.sub(r"[^A-Z0-9]", "", (key or "").upper()) return hashlib.sha256(f"{email}:{k}".encode()).hexdigest() def new_recovery_key(email): """Create (or replace) the recovery key for `email` and return it in XXXX-XXXX-XXXX-XXXX form.""" email = (email or "").strip().lower() raw = "".join(secrets.choice(KEY_ALPHABET) for _ in range(16)) key = "-".join(raw[i:i + 4] for i in range(0, 16, 4)) with _pw_db() as con: con.execute("UPDATE users SET recovery_hash=? WHERE email=?", (_key_hash(email, key), email)) request_sync() return key def has_recovery_key(email): with _pw_db() as con: row = con.execute("SELECT recovery_hash FROM users WHERE email=?", ((email or "").lower(),)).fetchone() return bool(row and row[0]) def key_html(key, intro="Your recovery key"): return (f'
🔑 {intro}:' f'
{key}
' "Save it somewhere safe (screenshot or notes). It is the only way to reset your password yourself " "if you forget it. It is shown only once.
") def reset_with_key(email, key, new_pw, confirm): """Returns (user, new_key, error).""" email = (email or "").strip().lower() if not _EMAIL_RE.match(email): return None, None, "Please enter your account email." if not (key or "").strip(): return None, None, "Please enter your recovery key." if len(new_pw or "") < 8: return None, None, "New password must be at least 8 characters." if new_pw != confirm: return None, None, "Passwords do not match." now = time.time() with _pw_db() as con: con.execute("DELETE FROM reset_fails WHERE ts < ?", (now - 3600,)) fails = con.execute("SELECT COUNT(*) FROM reset_fails WHERE email=?", (email,)).fetchone()[0] if fails >= MAX_KEY_FAILS: return None, None, "Too many wrong attempts. Please wait an hour, or ask the admin to reset it." row = con.execute("SELECT name, active, recovery_hash FROM users WHERE email=?", (email,)).fetchone() if not row or not row[2] or not hmac.compare_digest(row[2], _key_hash(email, key)): con.execute("INSERT INTO reset_fails (email, ts) VALUES (?,?)", (email, now)) return None, None, "Email or recovery key is not correct." if row[1] == 0: return None, None, "This account has been disabled. Please contact CS Software Solutions." salt = secrets.token_hex(16) con.execute("UPDATE users SET salt=?, pw=?, last_login=CURRENT_TIMESTAMP WHERE email=?", (salt, _hash(new_pw, salt), email)) con.execute("DELETE FROM reset_fails WHERE email=?", (email,)) con.execute("UPDATE reset_requests SET status='done' WHERE email=? AND status='open'", (email,)) return {"email": email, "name": row[0]}, new_recovery_key(email), None def change_password(user, old_pw, new_pw, confirm): """Signed-in user changes their password. Returns (ok, message).""" if not user: return False, "Please sign in first." _, err = check_user(user["email"], old_pw) if err: return False, "Current password is not correct." if len(new_pw or "") < 8: return False, "New password must be at least 8 characters." if new_pw != confirm: return False, "New passwords do not match." salt = secrets.token_hex(16) with _db() as con: con.execute("UPDATE users SET salt=?, pw=? WHERE email=?", (salt, _hash(new_pw, salt), user["email"])) request_sync() return True, "Password changed." def request_admin_reset(email, contact): """User without a recovery key asks the admin for a reset. Neutral answer either way.""" email, contact = (email or "").strip().lower(), (contact or "").strip()[:120] if not _EMAIL_RE.match(email): return False, "Please enter your account email." if not contact: return False, "Please add a WhatsApp / phone number so the admin can send you a temporary password." with _pw_db() as con: known = con.execute("SELECT 1 FROM users WHERE email=?", (email,)).fetchone() dup = con.execute("SELECT 1 FROM reset_requests WHERE email=? AND status='open'", (email,)).fetchone() if known and not dup: con.execute("INSERT INTO reset_requests (email, contact) VALUES (?,?)", (email, contact)) request_sync() return True, ("Request sent. If an account exists for this email, the admin will contact you on " f"{html.escape(contact)} with a temporary password (usually within a few hours).") RESET_REQ_COLS = ["When (UTC)", "Email", "Contact", "Status"] def admin_reset_requests(user): _need_admin(user) with _pw_db() as con: rows = con.execute("SELECT ts, email, contact, status FROM reset_requests " "ORDER BY status='done', id DESC LIMIT 100").fetchall() return [[(r[0] or "")[:16], r[1], r[2], r[3]] for r in rows] # --------------------------------------------------------------------------- # Forgot password by email: 6-digit code (15 min, 5 tries, max 3 codes per hour). Works only when # email is configured (mailer.ENABLED); the recovery key and admin request above always work. # --------------------------------------------------------------------------- RESET_MINUTES = 15 def _code_db(): con = _pw_db() con.execute("CREATE TABLE IF NOT EXISTS reset_codes (email TEXT, code_hash TEXT, expires REAL, " "attempts INTEGER DEFAULT 0, created REAL)") return con # Sign-up with a confirmed email: the account is created only after the 6-digit code sent to the address is entered # (anyone could otherwise sign up with someone else's address). Off when email is not set up (sign-up then works as # before) or with SIGNUP_VERIFY=0. SIGNUP_VERIFY = os.getenv("SIGNUP_VERIFY", "1") != "0" def _signup_db(): con = _pw_db() con.execute("CREATE TABLE IF NOT EXISTS signup_codes (email TEXT, name TEXT, salt TEXT, pw TEXT, code_hash TEXT, " "expires REAL, attempts INTEGER DEFAULT 0, created REAL)") return con def verify_signups(): import mailer as M return SIGNUP_VERIFY and M.ENABLED def start_signup(name, email, password, confirm): """Checks the form and emails a code. Returns (ok, message); nothing is created yet.""" import mailer as M name, email = (name or "").strip(), (email or "").strip().lower() if not name: return False, "Please enter your name." if not _EMAIL_RE.match(email): return False, "Please enter a valid email address." if len(password or "") < 8: return False, "Password must be at least 8 characters." if password != confirm: return False, "Passwords do not match." now = time.time() with _signup_db() as con: if con.execute("SELECT 1 FROM users WHERE email=?", (email,)).fetchone(): return False, "An account with this email already exists. Please log in." if con.execute("SELECT COUNT(*) FROM signup_codes WHERE email=? AND created > ?", (email, now - 3600)).fetchone()[0] >= 3: return False, "Too many sign-up codes for this address. Please wait an hour and try again." code = f"{secrets.randbelow(10 ** 6):06d}" salt = secrets.token_hex(16) con.execute("INSERT INTO signup_codes (email, name, salt, pw, code_hash, expires, created) VALUES " "(?,?,?,?,?,?,?)", (email, name, salt, _hash(password, salt), hashlib.sha256(f"signup:{email}:{code}".encode()).hexdigest(), now + RESET_MINUTES * 60, now)) con.execute("DELETE FROM signup_codes WHERE expires < ?", (now - 86400,)) ok, _ = M.send_signup_code(email, name, code, RESET_MINUTES) if not ok: return False, "We could not send the email right now. Please try again in a few minutes." return True, (f"We have sent a 6-digit code to {html.escape(email)}. Enter it below to create your " f"account (it expires in {RESET_MINUTES} minutes; check your spam folder too).") def confirm_signup(email, code): """Returns (user, error). Creates the account from the details given at start_signup.""" email, code = (email or "").strip().lower(), (code or "").strip() now = time.time() with _signup_db() as con: row = con.execute("SELECT rowid, name, salt, pw, code_hash, attempts FROM signup_codes WHERE email=? AND " "expires > ? ORDER BY created DESC LIMIT 1", (email, now)).fetchone() if not row: return None, "The code has expired or was not requested. Press Create account again for a new code." if row[5] >= 5: return None, "Too many wrong attempts. Press Create account again for a new code." if not hmac.compare_digest(row[4], hashlib.sha256(f"signup:{email}:{code}".encode()).hexdigest()): con.execute("UPDATE signup_codes SET attempts = attempts + 1 WHERE rowid=?", (row[0],)) return None, "That code is not correct." if con.execute("SELECT 1 FROM users WHERE email=?", (email,)).fetchone(): return None, "An account with this email already exists. Please log in." con.execute("INSERT INTO users (email, name, salt, pw, last_login) VALUES (?,?,?,?,CURRENT_TIMESTAMP)", (email, row[1], row[2], row[3])) con.execute("DELETE FROM signup_codes WHERE email=?", (email,)) request_sync() return {"email": email, "name": row[1]}, None def request_reset(email): """Sends a code. Same neutral message whether or not the email has an account.""" import mailer as M email = (email or "").strip().lower() if not M.ENABLED: return False, "Email reset is not available right now — use your recovery key or ask the admin below." if not _EMAIL_RE.match(email): return False, "Please enter a valid email address." neutral = (f"If an account exists for {html.escape(email)}, a 6-digit code has been sent. It expires in " f"{RESET_MINUTES} minutes — check your spam folder too.") now = time.time() with _code_db() as con: row = con.execute("SELECT name, active FROM users WHERE email=?", (email,)).fetchone() recent = con.execute("SELECT COUNT(*) FROM reset_codes WHERE email=? AND created > ?", (email, now - 3600)).fetchone()[0] if not row or row[1] == 0: return True, neutral if recent >= 3: return False, "Too many reset requests. Please wait an hour and try again." code = f"{secrets.randbelow(10 ** 6):06d}" con.execute("INSERT INTO reset_codes (email, code_hash, expires, created) VALUES (?,?,?,?)", (email, hashlib.sha256(f"{email}:{code}".encode()).hexdigest(), now + RESET_MINUTES * 60, now)) con.execute("DELETE FROM reset_codes WHERE expires < ?", (now - 86400,)) ok, _ = M.send_reset_code(email, row[0], code, RESET_MINUTES) if not ok: return False, "We could not send the email right now — use your recovery key or ask the admin below." return True, neutral def confirm_reset(email, code, new_pw, confirm): """Returns (user, error).""" email, code = (email or "").strip().lower(), (code or "").strip() if len(new_pw or "") < 8: return None, "New password must be at least 8 characters." if new_pw != confirm: return None, "Passwords do not match." now = time.time() with _code_db() as con: row = con.execute("SELECT rowid, code_hash, attempts FROM reset_codes WHERE email=? AND expires > ? " "ORDER BY created DESC LIMIT 1", (email, now)).fetchone() if not row: return None, "The code has expired or was not requested. Please request a new code." if row[2] >= 5: return None, "Too many wrong attempts. Please request a new code." if not hmac.compare_digest(row[1], hashlib.sha256(f"{email}:{code}".encode()).hexdigest()): con.execute("UPDATE reset_codes SET attempts = attempts + 1 WHERE rowid=?", (row[0],)) return None, "That code is not correct." salt = secrets.token_hex(16) con.execute("UPDATE users SET salt=?, pw=?, last_login=CURRENT_TIMESTAMP WHERE email=?", (salt, _hash(new_pw, salt), email)) con.execute("DELETE FROM reset_codes WHERE email=?", (email,)) con.execute("UPDATE reset_requests SET status='done' WHERE email=? AND status='open'", (email,)) u = con.execute("SELECT name, active FROM users WHERE email=?", (email,)).fetchone() request_sync() if not u or u[1] == 0: return None, "This account is disabled." return {"email": email, "name": u[0]}, None def reset_any(email, code_or_key, new_pw, confirm): """One form for both: a 6-digit number is an email code, anything else a recovery key. Returns (user, new_key_or_None, error).""" if re.fullmatch(r"\s*\d{6}\s*", code_or_key or ""): user, err = confirm_reset(email, code_or_key, new_pw, confirm) return user, None, err return reset_with_key(email, code_or_key, new_pw, confirm)