Create auth.py

src/auth.py

@@ -0,0 +1,393 @@
+# src/auth.py
+import os, re, time, secrets, uuid, sqlite3
+from datetime import datetime, timedelta
+import streamlit as st
+import bcrypt
+from email_validator import validate_email as _validate_email, EmailNotValidError
+
+from src.utils import get_connection
+
+# Config
+LOCKOUT_MINUTES = int(os.getenv("LOCKOUT_MINUTES", "15"))
+PASSWORD_EXPIRY_DAYS = int(os.getenv("PASSWORD_EXPIRY_DAYS", "90"))
+PASSWORD_HISTORY_SIZE = int(os.getenv("PASSWORD_HISTORY_SIZE", "3"))
+SESSION_TIMEOUT_MINUTES = int(os.getenv("SESSION_TIMEOUT_MINUTES", "30"))
+PEPPER = os.getenv("PEPPER", "")  # set this in HF Secrets
+
+# Password policy
+PASS_MIN_LEN = 8
+_PASS_UPPER = re.compile(r"[A-Z]")
+_PASS_LOWER = re.compile(r"[a-z]")
+_PASS_DIGIT = re.compile(r"\d")
+_PASS_SPECIAL = re.compile(r"[^A-Za-z0-9]")
+
+def password_policy_errors(pw: str) -> list[str]:
+    errs = []
+    if len(pw) < PASS_MIN_LEN: errs.append(f"Min length {PASS_MIN_LEN}")
+    if not _PASS_UPPER.search(pw): errs.append("At least 1 uppercase")
+    if not _PASS_LOWER.search(pw): errs.append("At least 1 lowercase")
+    if not _PASS_DIGIT.search(pw): errs.append("At least 1 digit")
+    if not _PASS_SPECIAL.search(pw): errs.append("At least 1 special char")
+    return errs
+
+def normalize_email(email: str) -> str:
+    return (email or "").strip().lower()
+
+def is_alnum(s: str) -> bool:
+    return bool(re.fullmatch(r"[A-Za-z0-9]+", s or ""))
+
+def is_valid_username(u: str) -> bool:
+    return bool(re.fullmatch(r"[A-Za-z0-9_]{3,32}", u or ""))
+
+# Hashing
+def hash_password(pw: str) -> str:
+    raw = (pw + PEPPER).encode("utf-8")
+    return "bcrypt$" + bcrypt.hashpw(raw, bcrypt.gensalt(rounds=12)).decode("utf-8")
+
+def verify_password(pw: str, stored: str) -> bool:
+    if not stored or not stored.startswith("bcrypt$"):
+        return False
+    hashed = stored[len("bcrypt$"):].encode("utf-8")
+    raw = (pw + PEPPER).encode("utf-8")
+    try:
+        return bcrypt.checkpw(raw, hashed)
+    except Exception:
+        return False
+
+# Schema
+def init_auth_schema():
+    with get_connection() as conn:
+        conn.execute("""
+        CREATE TABLE IF NOT EXISTS users (
+          id INTEGER PRIMARY KEY AUTOINCREMENT,
+          username TEXT NOT NULL UNIQUE,
+          email TEXT NOT NULL UNIQUE,
+          customer_id TEXT NOT NULL,
+          mobile TEXT,
+          role TEXT NOT NULL CHECK (role IN ('admin','user')),
+          password_hash TEXT NOT NULL,
+          password_algo TEXT NOT NULL DEFAULT 'bcrypt',
+          password_changed_at TEXT,
+          failed_attempts INTEGER NOT NULL DEFAULT 0,
+          locked_until TEXT,
+          is_active INTEGER NOT NULL DEFAULT 1,
+          last_login_at TEXT,
+          created_at TEXT NOT NULL,
+          updated_at TEXT NOT NULL
+        )""")
+        conn.execute("CREATE INDEX IF NOT EXISTS idx_users_email ON users(email)")
+
+        conn.execute("""
+        CREATE TABLE IF NOT EXISTS user_password_history (
+          id INTEGER PRIMARY KEY AUTOINCREMENT,
+          user_id INTEGER NOT NULL,
+          password_hash TEXT NOT NULL,
+          changed_at TEXT NOT NULL,
+          FOREIGN KEY(user_id) REFERENCES users(id) ON DELETE CASCADE
+        )""")
+        conn.execute("CREATE INDEX IF NOT EXISTS idx_uph_user ON user_password_history(user_id)")
+
+        conn.execute("""
+        CREATE TABLE IF NOT EXISTS login_audit (
+          id INTEGER PRIMARY KEY AUTOINCREMENT,
+          timestamp TEXT NOT NULL,
+          email TEXT NOT NULL,
+          user_id INTEGER,
+          result TEXT NOT NULL,   -- success | failure | locked
+          reason TEXT,
+          client_ip TEXT,
+          user_agent TEXT,
+          session_id TEXT
+        )""")
+        conn.commit()
+
+def bootstrap_admin_from_env():
+    with get_connection() as conn:
+        count = conn.execute("SELECT COUNT(*) FROM users").fetchone()[0]
+        if count > 0:
+            return
+        email = normalize_email(os.getenv("ADMIN_EMAIL", ""))
+        pw = os.getenv("ADMIN_PASSWORD", "")
+        name = os.getenv("ADMIN_NAME", "admin")
+        cust = os.getenv("ADMIN_CUSTOMER_ID", "ADMIN1")
+
+        if not email or not pw:
+            st.warning("Admin bootstrap: set ADMIN_EMAIL and ADMIN_PASSWORD for first-run admin.")
+            return
+        try:
+            _validate_email(email)
+        except EmailNotValidError:
+            st.warning("Admin bootstrap: invalid ADMIN_EMAIL.")
+            return
+        if not is_valid_username(name):
+            st.warning("Admin bootstrap: ADMIN_NAME must be 3–32 [A-Za-z0-9_].")
+            return
+        if not is_alnum(cust):
+            st.warning("Admin bootstrap: ADMIN_CUSTOMER_ID must be alphanumeric.")
+            return
+        errs = password_policy_errors(pw)
+        if errs:
+            st.warning("Admin bootstrap password policy: " + ", ".join(errs))
+            return
+
+        now = datetime.utcnow().isoformat(timespec="seconds")
+        pwh = hash_password(pw)
+        conn.execute("""
+            INSERT INTO users (username, email, customer_id, role, password_hash,
+                               password_algo, password_changed_at, created_at, updated_at, is_active)
+            VALUES (?, ?, ?, 'admin', ?, 'bcrypt', ?, ?, ?, 1)
+        """, (name, email, cust, pwh, now, now, now))
+        uid = conn.execute("SELECT id FROM users WHERE email=?", (email,)).fetchone()[0]
+        conn.execute("INSERT INTO user_password_history (user_id, password_hash, changed_at) VALUES (?, ?, ?)",
+                     (uid, pwh, now))
+        conn.commit()
+
+# Session helpers
+def set_session_user(row: sqlite3.Row):
+    st.session_state["_sid"] = st.session_state.get("_sid") or str(uuid.uuid4())
+    st.session_state["user"] = {
+        "id": row["id"],
+        "username": row["username"],
+        "email": row["email"],
+        "role": row["role"],
+        "customer_id": row["customer_id"],
+        "last_activity": time.time(),
+        "password_changed_at": row["password_changed_at"],
+    }
+
+def session_active() -> bool:
+    u = st.session_state.get("user")
+    if not u:
+        return False
+    idle = time.time() - u.get("last_activity", 0)
+    if idle > SESSION_TIMEOUT_MINUTES * 60:
+        st.session_state.pop("user", None)
+        return False
+    u["last_activity"] = time.time()
+    st.session_state["user"] = u
+    return True
+
+def session_countdown_widget():
+    u = st.session_state.get("user")
+    if not u:
+        return
+    idle = time.time() - u.get("last_activity", 0)
+    remain = max(0, SESSION_TIMEOUT_MINUTES*60 - int(idle))
+    mins, secs = divmod(remain, 60)
+    st.sidebar.caption(f"⏳ Session expires in {mins:02d}:{secs:02d}")
+
+# CAPTCHA
+def gen_captcha():
+    a, b = secrets.randbelow(8) + 2, secrets.randbelow(8) + 2
+    st.session_state["_captcha"] = (a, b, a + b)
+
+def check_captcha(ans: str) -> bool:
+    a, b, s = st.session_state.get("_captcha", (0,0,0))
+    try:
+        return int(ans) == s
+    except Exception:
+        return False
+
+# Queries
+def user_by_email(email: str) -> sqlite3.Row | None:
+    with get_connection() as conn:
+        conn.row_factory = sqlite3.Row
+        return conn.execute("SELECT * FROM users WHERE email=?", (email,)).fetchone()
+
+def email_exists(email: str) -> bool:
+    with get_connection() as conn:
+        return bool(conn.execute("SELECT 1 FROM users WHERE email=?", (email,)).fetchone())
+
+def username_exists(username: str) -> bool:
+    with get_connection() as conn:
+        return bool(conn.execute("SELECT 1 FROM users WHERE username=?", (username,)).fetchone())
+
+def within_lock(row: sqlite3.Row) -> bool:
+    val = row["locked_until"]
+    if not val: return False
+    try:
+        return datetime.utcnow() < datetime.fromisoformat(val)
+    except Exception:
+        return False
+
+def is_password_expired(row: sqlite3.Row) -> bool:
+    changed = row["password_changed_at"]
+    if not changed: return True
+    try:
+        return datetime.utcnow() - datetime.fromisoformat(changed) > timedelta(days=PASSWORD_EXPIRY_DAYS)
+    except Exception:
+        return True
+
+def record_login_audit(email: str, result: str, reason: str = "", user_id: int | None = None):
+    with get_connection() as conn:
+        conn.execute("""
+            INSERT INTO login_audit (timestamp, email, user_id, result, reason, client_ip, user_agent, session_id)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+        """, (
+            datetime.utcnow().isoformat(timespec="seconds"),
+            email, user_id, result, reason, None, None, st.session_state.get("_sid")
+        ))
+        conn.commit()
+
+def cannot_reuse_password(user_id: int, plaintext_pw: str) -> bool:
+    """Check plaintext+PEPPER against last N bcrypt hashes."""
+    with get_connection() as conn:
+        rows = conn.execute("""
+            SELECT password_hash FROM user_password_history
+            WHERE user_id=? ORDER BY changed_at DESC LIMIT ?
+        """, (user_id, PASSWORD_HISTORY_SIZE)).fetchall()
+    raw = (plaintext_pw + PEPPER).encode("utf-8")
+    for (stored,) in rows:
+        if not stored.startswith("bcrypt$"):
+            continue
+        if bcrypt.checkpw(raw, stored[len("bcrypt$"):].encode("utf-8")):
+            return True
+    return False
+
+# UI
+def render_login_signup_gate():
+    st.title("🔐 Sign in to Olist")
+    c1, c2 = st.columns(2)
+
+    # LOGIN
+    with c1:
+        st.subheader("Login")
+        with st.form("login_form"):
+            email = normalize_email(st.text_input("Email"))
+            pw = st.text_input("Password", type="password")
+            if "_captcha" not in st.session_state: gen_captcha()
+            a, b, _ = st.session_state.get("_captcha", (0,0,0))
+            cap = st.text_input(f"CAPTCHA: What is {a} + {b}?")
+            ok = st.form_submit_button("Login")
+        if ok:
+            if not check_captcha(cap):
+                st.error("CAPTCHA incorrect.")
+                gen_captcha()
+                return False
+            if not email or not pw:
+                st.error("Email and password required.")
+                return False
+            row = user_by_email(email)
+            if not row:
+                record_login_audit(email, "failure", "no_user", None)
+                st.error("Invalid credentials.")
+                gen_captcha(); return False
+            if int(row["is_active"]) != 1:
+                record_login_audit(email, "failure", "disabled", row["id"])
+                st.error("Account disabled. Contact admin.")
+                return False
+            if within_lock(row):
+                record_login_audit(email, "locked", "locked", row["id"])
+                st.error(f"Account locked until {row['locked_until']}.")
+                return False
+            if not verify_password(pw, row["password_hash"]):
+                with get_connection() as conn:
+                    fa = int(row["failed_attempts"]) + 1
+                    locked_until = None
+                    if fa >= 3:
+                        locked_until = (datetime.utcnow() + timedelta(minutes=LOCKOUT_MINUTES)).isoformat(timespec="seconds")
+                        fa = 0
+                    conn.execute(
+                        "UPDATE users SET failed_attempts=?, locked_until=?, updated_at=? WHERE id=?",
+                        (fa, locked_until, datetime.utcnow().isoformat(timespec="seconds"), row["id"])
+                    )
+                    conn.commit()
+                record_login_audit(email, "failure", "bad_password", row["id"])
+                if locked_until:
+                    st.error(f"Too many attempts. Locked until {locked_until}.")
+                else:
+                    st.error(f"Invalid credentials. Attempts left: {3 - (int(row['failed_attempts']) + 1)}")
+                gen_captcha(); return False
+
+                # not reached
+            # success
+            with get_connection() as conn:
+                conn.execute("UPDATE users SET failed_attempts=0, locked_until=NULL, last_login_at=?, updated_at=? WHERE id=?",
+                             (datetime.utcnow().isoformat(timespec="seconds"), datetime.utcnow().isoformat(timespec="seconds"), row["id"]))
+                conn.commit()
+            record_login_audit(email, "success", "", row["id"])
+            set_session_user(row)
+
+            if is_password_expired(row):
+                st.warning("Password expired. Please update.")
+                st.session_state["_force_change_pw"] = True
+                return False
+
+            return True
+
+    # SIGNUP
+    with c2:
+        st.subheader("Self-Signup")
+        with st.form("signup_form", clear_on_submit=True):
+            username = st.text_input("Username (3–32, letters/digits/_)")
+            email2 = normalize_email(st.text_input("Email"))
+            cust = st.text_input("Customer ID (alphanumeric)")
+            pw1 = st.text_input("Password", type="password")
+            pw2 = st.text_input("Confirm Password", type="password")
+            ok2 = st.form_submit_button("Create Account")
+        if ok2:
+            try:
+                _validate_email(email2)
+            except EmailNotValidError:
+                st.error("Invalid email format."); return False
+            if not is_valid_username(username):
+                st.error("Username must be 3–32 and letters/digits/_ only."); return False
+            if not is_alnum(cust):
+                st.error("Customer ID must be alphanumeric."); return False
+            if email_exists(email2):
+                st.error("Email already in use."); return False
+            if username_exists(username):
+                st.error("Username already in use."); return False
+            if pw1 != pw2:
+                st.error("Passwords do not match."); return False
+            errs = password_policy_errors(pw1)
+            if errs:
+                st.error("Password policy: " + ", ".join(errs)); return False
+
+            now = datetime.utcnow().isoformat(timespec="seconds")
+            pwh = hash_password(pw1)
+            with get_connection() as conn:
+                conn.execute("""
+                    INSERT INTO users (username, email, customer_id, role, password_hash,
+                                       password_algo, password_changed_at, created_at, updated_at, is_active)
+                    VALUES (?, ?, ?, 'user', ?, 'bcrypt', ?, ?, ?, 1)
+                """, (username, email2, cust, pwh, now, now, now))
+                uid = conn.execute("SELECT id FROM users WHERE email=?", (email2,)).fetchone()[0]
+                conn.execute("INSERT INTO user_password_history (user_id, password_hash, changed_at) VALUES (?, ?, ?)",
+                             (uid, pwh, now))
+                conn.commit()
+            st.success("Account created. Please log in.")
+            return False
+
+    return False
+
+def render_force_change_password():
+    st.subheader("Set a new password")
+    with st.form("change_pw", clear_on_submit=True):
+        p1 = st.text_input("New Password", type="password")
+        p2 = st.text_input("Confirm Password", type="password")
+        ok = st.form_submit_button("Update Password")
+    if ok:
+        errs = password_policy_errors(p1)
+        if p1 != p2:
+            st.error("Passwords do not match."); return False
+        if errs:
+            st.error("Password policy: " + ", ".join(errs)); return False
+        u = st.session_state.get("user")
+        if not u:
+            st.error("Session missing. Please log in again."); return False
+        if cannot_reuse_password(u["id"], p1):
+            st.error(f"Cannot reuse any of your last {PASSWORD_HISTORY_SIZE} passwords.")
+            return False
+        now = datetime.utcnow().isoformat(timespec="seconds")
+        pwh = hash_password(p1)
+        with get_connection() as conn:
+            conn.execute("UPDATE users SET password_hash=?, password_changed_at=?, updated_at=? WHERE id=?",
+                         (pwh, now, now, u["id"]))
+            conn.execute("INSERT INTO user_password_history (user_id, password_hash, changed_at) VALUES (?, ?, ?)",
+                         (u["id"], pwh, now))
+            conn.commit()
+        st.session_state.pop("_force_change_pw", None)
+        st.success("Password updated. Continue to the app.")
+        return True
+    return False