auth (1).py

"""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