"""OmniParse AI — Auth: bcrypt hashing, HttpOnly session cookies, password policy, HIBP check.""" import re, secrets, hashlib, bcrypt from datetime import datetime, timedelta, timezone from config import ( BCRYPT_ROUNDS, MIN_PASSWORD_LENGTH, REQUIRE_MIXED_CASE, REQUIRE_DIGIT, REQUIRE_SPECIAL, BANNED_PASSWORDS, SESSION_TTL_HOURS, COOKIE_NAME, COOKIE_SECURE, COOKIE_HTTPONLY, COOKIE_SAMESITE, ) from database import ( get_user_by_email, create_user as db_create_user, create_session, get_session, delete_session, get_user_by_id, ) # ── Password hashing ──────────────────────────────────────────────────────── def hash_password(plain: str) -> str: return bcrypt.hashpw(plain.encode("utf-8"), bcrypt.gensalt(rounds=BCRYPT_ROUNDS)).decode("utf-8") def verify_password(plain: str, hashed: str) -> bool: try: return bcrypt.checkpw(plain.encode("utf-8"), hashed.encode("utf-8")) except (ValueError, TypeError): return False # ── Password policy ───────────────────────────────────────────────────────── def _sha1(s: str) -> str: return hashlib.sha1(s.encode("utf-8")).hexdigest().upper() def check_password_strength(pw: str) -> list[str]: errs = [] if len(pw) < MIN_PASSWORD_LENGTH: errs.append(f"At least {MIN_PASSWORD_LENGTH} characters required.") if REQUIRE_MIXED_CASE and (pw.lower() == pw or pw.upper() == pw): errs.append("Must contain uppercase and lowercase letters.") if REQUIRE_DIGIT and not re.search(r"\d", pw): errs.append("Must contain at least one digit.") if REQUIRE_SPECIAL and not re.search(r'[!@#$%^&*()_+\-=\[\]{}|;:\'",.<>?/~`]', pw): errs.append("Must contain at least one special character (!@#$…).") if pw.lower() in BANNED_PASSWORDS: errs.append("Password is too common. Choose something stronger.") if re.search(r"(.)\1{4,}", pw): errs.append("Must not contain 5+ repeated characters.") return errs def check_password_leaked(pw: str) -> bool: """k-anonymity check via HIBP range API. Returns True if leaked.""" try: import requests s = _sha1(pw); prefix, suffix = s[:5], s[5:] resp = requests.get( f"https://api.pwnedpasswords.com/range/{prefix}", timeout=5, headers={"User-Agent": "OmniParse-AI"}, ) if resp.status_code != 200: return False return any(line.split(":")[0] == suffix for line in resp.text.splitlines()) except Exception: return False # fail open def validate_password_full(pw: str) -> tuple[bool, list[str]]: errs = check_password_strength(pw) if errs: return False, errs if check_password_leaked(pw): errs.append("This password appeared in known data breaches. Choose another.") return False, errs return True, [] # ── Session management ────────────────────────────────────────────────────── def generate_session_token() -> str: return secrets.token_urlsafe(48) def create_user_session(user_id: int) -> str: token = generate_session_token() expires = (datetime.now(timezone.utc) + timedelta(hours=SESSION_TTL_HOURS)).isoformat() create_session(user_id, token, expires) return token def validate_session_token(token: str | None) -> dict | None: if not token: return None s = get_session(token) return get_user_by_id(s["user_id"]) if s else None def set_session_cookie(response, token: str) -> None: response.set_cookie( key=COOKIE_NAME, value=token, httponly=COOKIE_HTTPONLY, secure=COOKIE_SECURE, samesite=COOKIE_SAMESITE, max_age=SESSION_TTL_HOURS * 3600, path="/", ) def clear_session_cookie(response) -> None: response.delete_cookie(key=COOKIE_NAME, path="/", httponly=COOKIE_HTTPONLY, secure=COOKIE_SECURE, samesite=COOKIE_SAMESITE) # ── Auth actions ──────────────────────────────────────────────────────────── def register_user(email: str, name: str, password: str) -> tuple[dict | None, str | None]: email = email.strip().lower() if not re.match(r"^[^@\s]+@[^@\s]+\.[^@\s]+$", email): return None, "Please enter a valid email address." name = name.strip() if len(name) < 2: return None, "Name must be at least 2 characters." if len(name) > 100: return None, "Name must be 100 characters or fewer." if not re.match(r"^[a-zA-ZÀ-ž\s\-']+$", name): return None, "Name contains invalid characters." valid, errs = validate_password_full(password) if not valid: return None, " ".join(errs) if get_user_by_email(email): return None, "An account with this email already exists." user = db_create_user(email, name, hash_password(password)) return (user, None) if user else (None, "Account creation failed. Try again.") def login_user(email: str, password: str) -> tuple[dict | None, str | None]: email = email.strip().lower() if not email or not password: return None, "Enter both email and password." user = get_user_by_email(email) # constant-time defense against user enumeration: if not user: bcrypt.checkpw(password.encode("utf-8"), bcrypt.gensalt(rounds=BCRYPT_ROUNDS)) return None, "Invalid email or password." if not verify_password(password, user["password_hash"]): return None, "Invalid email or password." return user, None