| """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, |
| ) |
|
|
| |
| 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 |
|
|
| |
| 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 |
|
|
| 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, [] |
|
|
| |
| 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) |
|
|
| |
| 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) |
| |
| 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 |
|
|
