"""Authentication routes: registration, login and logout. Routes stay thin — request handling, flashing and redirects only. Account creation and credential checks live in :mod:`app.auth.services`. """ from __future__ import annotations import os import uuid from flask import ( current_app, flash, jsonify, make_response, redirect, render_template, request, send_from_directory, session, url_for, ) from flask_login import current_user, login_required, login_user, logout_user from werkzeug.security import check_password_hash from werkzeug.utils import secure_filename from app.auth import auth_bp from app.auth.forms import ( ChangeEmailForm, ConfirmPasswordForm, DeviceRenameForm, ForgotPasswordForm, LoginForm, LoginOtpForm, OtpVerifyForm, PasswordChangeForm, ProfilePictureForm, ProfileUpdateForm, RegisterForm, ResendOtpForm, ResetPasswordForm, StudentVerificationForm, TrustedDeviceActionForm, ) from app.auth.forms import _PASSWORD_POLICY_MESSAGE from app.auth import registration from app.auth import security_service from app.auth import trusted_devices from app.auth import verification as verification_service from app.auth.password_reset import ( complete_password_reset, start_password_reset, verify_password_reset, ) from app.auth.otp_service import ( OtpRateLimited, audit_event, ensure_active_code, invalidate_codes, issue_verification_code, seconds_until_resend, verify_code, ) from app.auth.services import ( AuthEvent, authenticate, change_password, change_pending_email, get_user_by_email, is_account_locked, log_auth_event, mark_email_verified, register_failed_login, reset_login_failures, save_profile_picture, ) from app.extensions import db, login_manager # Shared client-IP resolver (also used by view tracking / admin audit) so the # security log lines can record the request IP without a second implementation. from app.marketplace.services import _client_ip from app.models import Conversation, Favorite, Listing, ListingStatus, User from app.models.base import utcnow from app.models.enums import VerificationPurpose # The login/registration gate verifies email ownership. Other purposes # (password reset, 2FA…) reuse the same framework with a different value. _EMAIL_PURPOSE = VerificationPurpose.EMAIL_VERIFICATION # Session keys for the email-verification handshake. The pending user is tracked # in the (signed) session so registration and the login gate share one flow; the # account stays unauthenticated until the emailed OTP is confirmed. _PENDING_KEY = "pending_verification_user_id" _ATTEMPTS_KEY = "otp_verify_attempts" # Verify-first registration: the pending TemporaryRegistration id (no user yet). _REG_TEMP_ID = "reg_temp_id" # Forgot-password handshake (three steps, unauthenticated). The account is never # logged in during recovery — a fresh login is required afterwards. _PW_RESET_EMAIL = "pw_reset_email" # step 1 → identifies the account _PW_RESET_UID = "pw_reset_uid" # step 2 → authorises the new-password step _PW_RESET_ATTEMPTS = "pw_reset_attempts" # Login-verification handshake (untrusted device). Credentials are already valid # at this point; the account stays unauthenticated until the emailed login code # is confirmed. Distinct session keys from the email-verification gate above so # the two flows never collide. _LOGIN_PURPOSE = VerificationPurpose.LOGIN_VERIFICATION _LOGIN_PENDING_KEY = "login_pending_user_id" _LOGIN_ATTEMPTS_KEY = "login_otp_attempts" _LOGIN_REMEMBER_KEY = "login_pending_remember" # carry "Remember me" across the OTP step _LOGIN_NEXT_KEY = "login_pending_next" # carry the post-login redirect target @login_manager.user_loader def load_user(user_id: str) -> User | None: """Reload a user from the session-stored id (Flask-Login requirement).""" return db.session.get(User, int(user_id)) def _safe_redirect_target() -> str: """Return a safe, in-app redirect target from the ``next`` parameter.""" target = request.args.get("next", "") if target.startswith("/") and not target.startswith("//"): return target return url_for("main.marketplace") # Profile picture upload policy: images only, generous but bounded size. _PROFILE_PICTURE_EXTENSIONS = {"png", "jpg", "jpeg", "webp"} _PROFILE_PICTURE_MAX_BYTES = 5 * 1024 * 1024 # 5 MB def _process_profile_picture(form: ProfilePictureForm, user: User) -> tuple[bool, str, str | None]: """Validate and store an uploaded profile picture. Returns ``(success, message, url)`` where ``url`` points at the newly saved image on success. All validation is server-side; the client-side checks are only a convenience. """ if not form.validate_on_submit(): # CSRF failure or the form's FileAllowed validator rejected the type. errors = form.profile_picture.errors or ["Please choose a valid image."] return False, errors[0], None file = form.profile_picture.data if not file or not file.filename: return False, "Please choose an image to upload.", None filename = secure_filename(file.filename) ext = filename.rsplit(".", 1)[-1].lower() if "." in filename else "" if ext not in _PROFILE_PICTURE_EXTENSIONS: return False, "Only PNG, JPG, JPEG or WEBP images are allowed.", None # Measure the stream without loading it all into memory twice. file.seek(0, os.SEEK_END) size = file.tell() file.seek(0) if size == 0: return False, "The selected file is empty.", None if size > _PROFILE_PICTURE_MAX_BYTES: return False, "Profile photo must be 5MB or smaller.", None folder = current_app.config["PROFILE_PICTURE_FOLDER"] folder.mkdir(parents=True, exist_ok=True) stored_name = f"{uuid.uuid4().hex}.{ext}" previous = user.profile.profile_picture if user.profile else None file.save(folder / stored_name) save_profile_picture(user, stored_name) # Clean up the old file so uploads don't accumulate as orphans. if previous and previous != stored_name: try: old_path = folder / previous if old_path.exists(): old_path.unlink() except OSError: pass # Non-fatal: the new picture is already saved. return ( True, "Profile photo updated successfully.", url_for("auth.profile_picture", filename=stored_name), ) def _pending_user() -> User | None: """The account currently awaiting email verification (from the session).""" uid = session.get(_PENDING_KEY) return db.session.get(User, uid) if uid else None def _begin_verification(user: User) -> None: """Start (or resume) the verification handshake for ``user``: remember them in the session and make sure a live code has been emailed.""" session[_PENDING_KEY] = user.id session.pop(_ATTEMPTS_KEY, None) ensure_active_code(user, _EMAIL_PURPOSE) def _login_pending_user() -> User | None: """The account that has passed the password check and is awaiting its login code (from the session). ``None`` when there is no live login handshake.""" uid = session.get(_LOGIN_PENDING_KEY) return db.session.get(User, uid) if uid else None def _clear_login_pending() -> None: """Forget the login-verification handshake state.""" for key in (_LOGIN_PENDING_KEY, _LOGIN_ATTEMPTS_KEY, _LOGIN_REMEMBER_KEY, _LOGIN_NEXT_KEY): session.pop(key, None) def _safe_login_next() -> str: """Post-login redirect target stashed at the start of the login handshake, validated to be in-app (never an open redirect).""" target = session.get(_LOGIN_NEXT_KEY, "") or "" if target.startswith("/") and not target.startswith("//"): return target return url_for("main.marketplace") def _begin_login_verification(user: User, *, remember: bool) -> None: """Start the login-OTP handshake for a user whose password is already valid but whose device isn't trusted: remember them (unauthenticated) in the session, stash their preferences, and email a login code if none is live.""" session[_LOGIN_PENDING_KEY] = user.id session[_LOGIN_REMEMBER_KEY] = bool(remember) session[_LOGIN_NEXT_KEY] = request.args.get("next", "") or "" session.pop(_LOGIN_ATTEMPTS_KEY, None) ensure_active_code( user, _LOGIN_PURPOSE, expiry_minutes=current_app.config.get("LOGIN_OTP_EXPIRY_MINUTES"), ) log_auth_event(AuthEvent.LOGIN_OTP_SENT, user=user, ip=_client_ip(request)) def _mask_email(email: str) -> str: """Partially obscure an email for display (``a****z@campus.edu``).""" local, _, domain = email.partition("@") if not domain: return email if len(local) <= 2: masked = local[0] + "*" else: masked = local[0] + "*" * (len(local) - 2) + local[-1] return f"{masked}@{domain}" @auth_bp.route("/register", methods=["GET", "POST"]) def register(): """Verify-first registration: validate, store a *pending* registration and email an OTP. No user/profile/session/cookie is created until the code is confirmed (PART 1), so the database holds zero unverified accounts.""" if current_user.is_authenticated: return redirect(url_for("main.marketplace")) form = RegisterForm() # Prefill from a live pending registration so navigating back keeps the # fields (password can't be prefilled — only its hash is stored). if request.method == "GET": pending = registration.get_pending(session.get(_REG_TEMP_ID)) if pending is not None: form.full_name.data = pending.full_name form.email.data = pending.email if form.validate_on_submit(): temp, err = registration.start_registration( form.full_name.data, form.email.data, form.password.data) if err == registration.ERR_EXISTS: flash("An account already exists with this email.", "danger") elif err == registration.ERR_SEND_FAILED: flash("We couldn't send your verification email. Please try again in a moment.", "danger") elif temp is not None: session[_REG_TEMP_ID] = temp.id flash("We've emailed you a 6-digit code. Enter it below to verify your email.", "info") return redirect(url_for("auth.register_verify")) return render_template("auth/register.html", form=form) @auth_bp.route("/register/verify", methods=["GET", "POST"]) def register_verify(): """Confirm the registration OTP. On success the verified user is created and logged in automatically (PART 1) — no credential re-entry.""" if current_user.is_authenticated: return redirect(url_for("main.marketplace")) temp = registration.get_pending(session.get(_REG_TEMP_ID)) if temp is None: session.pop(_REG_TEMP_ID, None) flash("Your verification session expired. Please register again.", "warning") return redirect(url_for("auth.register")) form = OtpVerifyForm() if form.validate_on_submit(): user, err = registration.verify_registration(temp, form.code.data) if user is not None: session.pop(_REG_TEMP_ID, None) login_user(user) # auto-login after verification user.last_login = utcnow() db.session.commit() return render_template("auth/verify_success.html") if err == registration.ERR_EXPIRED: session.pop(_REG_TEMP_ID, None) flash("Your verification session expired. Please register again.", "warning") return redirect(url_for("auth.register")) if err == registration.ERR_EXISTS: session.pop(_REG_TEMP_ID, None) flash("An account already exists with this email.", "danger") return redirect(url_for("auth.login")) if err == registration.ERR_LOCKED: flash("Too many incorrect attempts. Please resend a new code.", "danger") elif err == registration.ERR_CODE_EXPIRED: flash("That code has expired. Please resend a new one.", "danger") else: flash("Incorrect code. Please check your email and try again.", "danger") return render_template( "auth/register_verify.html", form=form, resend_form=ResendOtpForm(), change_form=ChangeEmailForm(email=temp.email), email=temp.email, masked_email=_mask_email(temp.email), resend_wait=registration.seconds_until_resend(temp), ) @auth_bp.route("/register/resend", methods=["POST"]) def register_resend(): temp = registration.get_pending(session.get(_REG_TEMP_ID)) if temp is None: flash("Your verification session expired. Please register again.", "warning") return redirect(url_for("auth.register")) if ResendOtpForm().validate_on_submit(): ok, err = registration.resend_code(temp) if ok: flash("A new verification code is on its way to your inbox.", "info") elif err == registration.ERR_RATE_LIMITED: wait = registration.seconds_until_resend(temp) flash(f"Please wait {wait} seconds before requesting another code." if wait else "You've requested too many codes. Please try again later.", "warning") elif err == registration.ERR_SEND_FAILED: flash("We couldn't send your verification email. Please try again in a moment.", "danger") elif err == registration.ERR_EXPIRED: session.pop(_REG_TEMP_ID, None) flash("Your verification session expired. Please register again.", "warning") return redirect(url_for("auth.register")) return redirect(url_for("auth.register_verify")) @auth_bp.route("/register/change-email", methods=["POST"]) def register_change_email(): """Fix a mistyped email before verifying, then resend — no need to re-enter name or password (PART 4).""" temp = registration.get_pending(session.get(_REG_TEMP_ID)) if temp is None: flash("Your verification session expired. Please register again.", "warning") return redirect(url_for("auth.register")) change_form = ChangeEmailForm() if change_form.validate_on_submit(): ok, err = registration.change_email(temp, change_form.email.data) if ok: flash(f"Email updated. We've sent a fresh code to {temp.email}.", "success") elif err == registration.ERR_EXISTS: flash("An account already exists with this email.", "danger") elif err == registration.ERR_SEND_FAILED: flash("We couldn't send your verification email. Please try again in a moment.", "danger") elif err == registration.ERR_EXPIRED: session.pop(_REG_TEMP_ID, None) flash("Your verification session expired. Please register again.", "warning") return redirect(url_for("auth.register")) else: flash("Please enter a valid email address.", "danger") return redirect(url_for("auth.register_verify")) @auth_bp.route("/login", methods=["GET", "POST"]) def login(): if current_user.is_authenticated: return redirect(url_for("main.marketplace")) form = LoginForm() if form.validate_on_submit(): ip = _client_ip(request) # Temporary lockout check (Security_Design §7). Looking the account up by # email first lets us track failures and enforce the lock without ever # revealing whether the address exists (the message stays uniform). account = get_user_by_email(form.email.data) if account is not None and is_account_locked(account): log_auth_event("login_blocked_locked", user=account, ip=ip) flash( "Too many failed attempts. Please try again later or reset your password.", "danger", ) return render_template("auth/login.html", form=form) user = authenticate(form.email.data, form.password.data) if user is None: if account is not None: # real account, wrong password → count it register_failed_login(account, ip=ip) else: # unknown email → audit without a user log_auth_event("login_failed", ip=ip, reason="unknown_email") flash("Invalid email or password.", "danger") elif not user.is_verified: # Credentials valid but email never verified: reset the failure # counter and route to the existing email-verification gate. reset_login_failures(user) _begin_verification(user) flash( "Your email address hasn't been verified yet. We've sent you a " "code — enter it below to finish signing in.", "warning", ) return redirect(url_for("auth.verify_email")) else: # Password is correct: clear any failed-attempt state. reset_login_failures(user) # The OTP gate fires when the global flag is on OR the user has opted # into Email-OTP two-factor — either way, only on an untrusted device. otp_on = current_app.config.get("LOGIN_OTP_ENABLED", True) or user.two_factor_enabled is_trusted = trusted_devices.device_is_trusted(user, request) if otp_on and not is_trusted: # Untrusted device → require an emailed login code before the # session is established. The user is NOT logged in yet. _begin_login_verification(user, remember=form.remember.data) flash("We've sent a verification code to your email.", "info") return redirect(url_for("auth.login_verify")) # Trusted device (or OTP disabled) → sign in immediately. _finalize_login(user, ip=ip, method=("trusted_device" if is_trusted else "password")) login_user(user, remember=form.remember.data) db.session.commit() flash("Welcome back!", "success") return redirect(_safe_redirect_target()) return render_template("auth/login.html", form=form) def _finalize_login(user: User, *, ip: str | None, method: str, event: str = None) -> None: """Shared post-authentication bookkeeping: stamp last_login, run the suspicious-context check (email + audit on a new device/IP), and record the success event with its auth method. Commits are done by the caller.""" event = event or AuthEvent.LOGIN_SUCCESS # Detect a new browser/device/IP *before* recording this login (so it isn't # compared against itself). Trusted-device sign-ins are inherently known. flags = [] if method == "trusted_device" else security_service.detect_new_context(user, request) user.last_login = utcnow() if flags: # Only an *unrecognized* sign-in warns + emails — routine and trusted # logins stay silent (no per-login email spam). log_auth_event(AuthEvent.SUSPICIOUS_LOGIN, user=user, ip=ip, flags=",".join(flags)) flash("We noticed a sign-in from a new device or location. If this wasn't " "you, review your security settings.", "warning") security_service.notify_new_login(user, request, flags=flags) log_auth_event(event, user=user, ip=ip, method=method) @auth_bp.route("/login/verify", methods=["GET", "POST"]) def login_verify(): """Confirm the emailed login code for an untrusted device. Reached only after a successful password check (the pending user id lives in the session). On success the user is logged in, the device may be trusted for the configured window, and they land on their originally requested page. Wrong-code attempts are capped per issued code (brute-force guard).""" if current_user.is_authenticated: return redirect(url_for("main.marketplace")) user = _login_pending_user() if user is None: flash("Please log in to continue.", "info") return redirect(url_for("auth.login")) form = LoginOtpForm() if form.validate_on_submit(): ip = _client_ip(request) max_attempts = current_app.config.get("OTP_MAX_VERIFY_ATTEMPTS", 5) attempts = session.get(_LOGIN_ATTEMPTS_KEY, 0) + 1 session[_LOGIN_ATTEMPTS_KEY] = attempts ok, error = verify_code(user, _LOGIN_PURPOSE, form.code.data) if ok: remember = bool(session.get(_LOGIN_REMEMBER_KEY, False)) trust = bool(form.trust_device.data) next_target = _safe_login_next() _finalize_login(user, ip=ip, method="email_otp", event=AuthEvent.LOGIN_OTP_SUCCESS) login_user(user, remember=remember) db.session.commit() _clear_login_pending() response = make_response(redirect(next_target)) if trust: # Create the trusted-device record and set its secure cookie so # future sign-ins on this browser skip the OTP step. trusted_devices.remember_device(user, response, request) flash("Welcome back! You're now signed in.", "success") return response log_auth_event(AuthEvent.LOGIN_OTP_FAILURE, user=user, ip=ip, attempts=attempts) if attempts >= max_attempts: invalidate_codes(user, _LOGIN_PURPOSE) audit_event("locked", user, _LOGIN_PURPOSE, attempts=attempts) session[_LOGIN_ATTEMPTS_KEY] = 0 flash("Too many incorrect attempts. That code is now disabled — " "please request a new one.", "danger") else: flash(error, "danger") return render_template( "auth/login_verify.html", form=form, resend_form=ResendOtpForm(), email=user.email, masked_email=_mask_email(user.email), resend_wait=seconds_until_resend(user, _LOGIN_PURPOSE), ) @auth_bp.route("/login/resend", methods=["POST"]) def login_resend(): """Resend the login code (cooldown + hourly cap enforced by the framework).""" user = _login_pending_user() if user is None: flash("Please log in to continue.", "info") return redirect(url_for("auth.login")) if ResendOtpForm().validate_on_submit(): try: issue_verification_code( user, _LOGIN_PURPOSE, event="resent", expiry_minutes=current_app.config.get("LOGIN_OTP_EXPIRY_MINUTES"), ) session[_LOGIN_ATTEMPTS_KEY] = 0 log_auth_event(AuthEvent.LOGIN_OTP_RESEND, user=user, ip=_client_ip(request)) flash("A new verification code is on its way to your inbox.", "info") except OtpRateLimited as exc: flash(str(exc), "warning") return redirect(url_for("auth.login_verify")) @auth_bp.route("/verify-email", methods=["GET", "POST"]) def verify_email(): """Confirm the emailed OTP. Shared by registration and the login gate. On success the account is verified and logged in automatically — no need to re-enter credentials. Wrong-code attempts are capped per issued code.""" user = _pending_user() if user is None: flash("Please log in to continue.", "info") return redirect(url_for("auth.login")) if user.is_verified: session.pop(_PENDING_KEY, None) flash("Your email is already verified. Please log in.", "info") return redirect(url_for("auth.login")) form = OtpVerifyForm() if form.validate_on_submit(): max_attempts = current_app.config.get("OTP_MAX_VERIFY_ATTEMPTS", 5) attempts = session.get(_ATTEMPTS_KEY, 0) + 1 session[_ATTEMPTS_KEY] = attempts ok, error = verify_code(user, _EMAIL_PURPOSE, form.code.data) if ok: # Framework confirmed the code; apply the email-verification effect. mark_email_verified(user) session.pop(_PENDING_KEY, None) session.pop(_ATTEMPTS_KEY, None) login_user(user) # auto-login exactly as before user.last_login = utcnow() db.session.commit() # Show a success page (with the login already applied) rather than # bouncing straight to the marketplace. return render_template("auth/verify_success.html") if attempts >= max_attempts: invalidate_codes(user, _EMAIL_PURPOSE) audit_event("locked", user, _EMAIL_PURPOSE, attempts=attempts) session[_ATTEMPTS_KEY] = 0 flash("Too many incorrect attempts. That code is now disabled — " "please request a new one.", "danger") else: flash(error, "danger") return render_template( "auth/verify_email.html", form=form, resend_form=ResendOtpForm(), change_form=ChangeEmailForm(email=user.email), email=user.email, masked_email=_mask_email(user.email), resend_wait=seconds_until_resend(user, _EMAIL_PURPOSE), ) @auth_bp.route("/verify-email/resend", methods=["POST"]) def resend_otp(): user = _pending_user() if user is None: flash("Please log in to continue.", "info") return redirect(url_for("auth.login")) if user.is_verified: session.pop(_PENDING_KEY, None) return redirect(url_for("auth.login")) if ResendOtpForm().validate_on_submit(): try: # enforces cooldown + hourly cap; labelled "resent" in the audit log issue_verification_code(user, _EMAIL_PURPOSE, event="resent") session[_ATTEMPTS_KEY] = 0 flash("A new verification code is on its way to your inbox.", "info") except OtpRateLimited as exc: flash(str(exc), "warning") return redirect(url_for("auth.verify_email")) @auth_bp.route("/verify-email/change", methods=["POST"]) def change_email(): """Optional: correct a mistyped email before verifying, then re-send.""" user = _pending_user() if user is None: flash("Please log in to continue.", "info") return redirect(url_for("auth.login")) if user.is_verified: session.pop(_PENDING_KEY, None) return redirect(url_for("auth.login")) change_form = ChangeEmailForm() if change_form.validate_on_submit(): if change_pending_email(user, change_form.email.data): session[_ATTEMPTS_KEY] = 0 invalidate_codes(user, _EMAIL_PURPOSE) try: issue_verification_code(user, _EMAIL_PURPOSE, enforce_cooldown=False) flash(f"Email updated. We've sent a fresh code to {user.email}.", "success") except OtpRateLimited as exc: flash(f"Email updated, but {str(exc)[0].lower()}{str(exc)[1:]}", "warning") else: flash("That email address is already in use by another account.", "danger") else: flash("Please enter a valid email address.", "danger") return redirect(url_for("auth.verify_email")) # --- Forgot password (reuses the PASSWORD_RESET verification purpose) ---------- @auth_bp.route("/forgot-password", methods=["GET", "POST"]) def forgot_password(): """Step 1: identify the account and issue a reset code. Always shows the same generic confirmation so the response never reveals whether an email is registered.""" if current_user.is_authenticated: return redirect(url_for("main.marketplace")) form = ForgotPasswordForm(email=session.get(_PW_RESET_EMAIL) or None) if form.validate_on_submit(): start_password_reset(form.email.data) # all logic in the service session[_PW_RESET_EMAIL] = form.email.data.strip().lower() session.pop(_PW_RESET_UID, None) session.pop(_PW_RESET_ATTEMPTS, None) flash("If an account exists for that email, we've sent a password reset code.", "info") return redirect(url_for("auth.reset_verify")) return render_template("auth/forgot_password.html", form=form) @auth_bp.route("/reset-password/verify", methods=["GET", "POST"]) def reset_verify(): """Step 2: confirm the emailed reset code (single-use) and authorise step 3.""" if current_user.is_authenticated: return redirect(url_for("main.marketplace")) email = session.get(_PW_RESET_EMAIL) if not email: return redirect(url_for("auth.forgot_password")) form = OtpVerifyForm() # reuse the generic 6-digit code form if form.validate_on_submit(): max_attempts = current_app.config.get("OTP_MAX_VERIFY_ATTEMPTS", 5) attempts = session.get(_PW_RESET_ATTEMPTS, 0) + 1 session[_PW_RESET_ATTEMPTS] = attempts user = db.session.scalar(db.select(User).filter_by(email=email).limit(1)) ok = False if user is not None: ok, _ = verify_password_reset(user, form.code.data) if ok: session[_PW_RESET_UID] = user.id session.pop(_PW_RESET_ATTEMPTS, None) return redirect(url_for("auth.reset_password")) # Uniform failure message (never reveals whether the account exists). if user is not None and attempts >= max_attempts: invalidate_codes(user, VerificationPurpose.PASSWORD_RESET) audit_event("locked", user, VerificationPurpose.PASSWORD_RESET, attempts=attempts) session[_PW_RESET_ATTEMPTS] = 0 flash("Too many incorrect attempts. Please request a new reset code.", "danger") else: flash("That code is invalid or has expired. Request a new one if you need it.", "danger") return render_template( "auth/reset_verify.html", form=form, masked_email=_mask_email(email) ) @auth_bp.route("/reset-password", methods=["GET", "POST"]) def reset_password(): """Step 3: set the new password. Requires a verified reset code (step 2).""" if current_user.is_authenticated: return redirect(url_for("main.marketplace")) uid = session.get(_PW_RESET_UID) if not uid: flash("Please verify your reset code first.", "info") return redirect(url_for("auth.forgot_password")) user = db.session.get(User, uid) if user is None: session.pop(_PW_RESET_UID, None) return redirect(url_for("auth.forgot_password")) form = ResetPasswordForm() if form.validate_on_submit(): ok, error = complete_password_reset(user, form.new_password.data) if ok: # No auto-login: a fresh sign-in is required. The reset already # revoked every trusted device (in the service); drop this browser's # now-defunct trusted-device cookie too. session.pop(_PW_RESET_UID, None) session.pop(_PW_RESET_EMAIL, None) response = make_response(render_template("auth/reset_success.html")) trusted_devices.clear_cookie(response) return response flash(error, "danger") return render_template("auth/reset_password.html", form=form) @auth_bp.route("/profile/picture/") def profile_picture(filename: str): return send_from_directory( current_app.config["PROFILE_PICTURE_FOLDER"], filename, as_attachment=False, ) @auth_bp.route("/profile", methods=["GET", "POST"]) @login_required def profile(): profile = current_user.profile profile_data = { "full_name": profile.full_name if profile else current_user.email, "profile_picture": profile.profile_picture if profile else None, "phone_number": profile.phone_number if profile else "", "bio": profile.bio if profile else "", "preferred_meeting_location": getattr(profile, "preferred_meeting_location", ""), "location": getattr(profile, "location", "Campus"), } profile_update_form = ProfileUpdateForm( full_name=profile_data["full_name"], email=current_user.email, bio=profile_data["bio"], preferred_meeting_location=profile_data["preferred_meeting_location"], ) profile_picture_form = ProfilePictureForm() verification_form = StudentVerificationForm() if request.method == "GET" and profile: verification_form.mobile_number.data = profile.phone_number verification_form.university_name.data = getattr(profile, "university_name", None) verification_form.university_student_id.data = getattr(profile, "university_student_id", None) if request.method == "POST": if "submit_verification" in request.form: from app.auth.verification import VerificationError, submit_verification if verification_form.validate_on_submit(): try: submit_verification( current_user, verification_form.mobile_number.data, verification_form.university_name.data, verification_form.university_student_id.data, ) flash("Your verification request has been submitted for review.", "success") except VerificationError as exc: flash(str(exc), "danger") else: flash("Please complete all verification fields.", "danger") return redirect(url_for("auth.profile")) if "save_profile" in request.form: if profile_update_form.validate_on_submit(): from app.auth.services import update_profile update_profile( current_user, full_name=profile_update_form.full_name.data, bio=profile_update_form.bio.data, preferred_meeting_location=profile_update_form.preferred_meeting_location.data, ) flash("Profile updated successfully.", "success") return redirect(url_for("auth.profile")) else: # Profile picture upload. The avatar form auto-submits on file # selection, so it carries no button field — detect it as the # non-"save_profile" POST. Supports both AJAX (fetch) and a plain # form fallback. wants_json = request.headers.get("X-Requested-With") == "fetch" success, message, picture_url = _process_profile_picture( profile_picture_form, current_user ) if wants_json: return ( jsonify(success=success, message=message, url=picture_url), 200 if success else 400, ) flash(message, "success" if success else "danger") return redirect(url_for("auth.profile")) listing_count = db.session.scalar( db.select(db.func.count()) .select_from(Listing) .where(Listing.seller_id == current_user.id, Listing.deleted_at.is_(None)) ) or 0 conversation_count = db.session.scalar( db.select(db.func.count()) .select_from(Conversation) .where( db.or_( Conversation.buyer_id == current_user.id, Conversation.seller_id == current_user.id, ) ) ) or 0 saved_count = db.session.scalar( db.select(db.func.count()) .select_from(Favorite) .where(Favorite.user_id == current_user.id) ) or 0 sold_count = db.session.scalar( db.select(db.func.count()) .select_from(Listing) .where( Listing.seller_id == current_user.id, Listing.status == ListingStatus.SOLD, Listing.deleted_at.is_(None), ) ) or 0 # Premium status card. Retire lapsed subscriptions on access so the card # reflects reality (no scheduled job), then read the live state. from app.premium import services as premium_services premium_services.cleanup_expired_premium_users() active_subscription = premium_services.get_active_subscription(current_user) latest_subscription = ( active_subscription if active_subscription is not None else premium_services.get_latest_subscription(current_user) ) pending_request = ( None if active_subscription is not None else premium_services.get_pending_request(current_user) ) return render_template( "auth/profile.html", profile=profile_data, listing_count=listing_count, conversation_count=conversation_count, saved_count=saved_count, sold_count=sold_count, profile_update_form=profile_update_form, profile_picture_form=profile_picture_form, verification_form=verification_form, verification_status=current_user.student_verification_status.value, verification_history=verification_service.get_history_timeline(current_user, include_moderation=False), # Draft "edit verification" mode is a pure UI state (no server change, # no badge loss) — a verified student toggles it via ?verify_edit=1. verify_edit=(request.args.get("verify_edit") == "1"), active_subscription=active_subscription, latest_subscription=latest_subscription, pending_request=pending_request, premium_plans=premium_services.PLAN_CATALOG, ) @auth_bp.route("/profile/resend-verification", methods=["POST"]) @login_required def resend_verification(): """Resend a verification code from the profile page. Defensive: a signed-in account is normally already verified (unverified accounts can't log in). Kept for accounts left unverified by other means so they can self-serve a fresh code.""" if current_user.is_verified: flash("Your email address is already verified.", "info") return redirect(url_for("auth.profile")) try: issue_verification_code(current_user, _EMAIL_PURPOSE) session[_PENDING_KEY] = current_user.id flash("We've sent a verification code to your email.", "info") return redirect(url_for("auth.verify_email")) except OtpRateLimited as exc: flash(str(exc), "warning") return redirect(url_for("auth.profile")) @auth_bp.route("/seller/") def seller_profile(public_id: str): """Trust-system public seller profile URL — delegates to the canonical public-profile view so there is a single implementation (and ``url_for`` still builds ``/users/`` for the existing profile links).""" return public_profile(public_id) @auth_bp.route("/users/") def public_profile(public_id: str): """Public marketplace profile for a user. Shows only information appropriate for a public marketplace listing page (name, campus, bio, marketplace stats and active listings) — never email, phone, security or other private account data. The owner is redirected to their own editable profile so there is a single editing experience. """ from app.marketplace.services import ( get_public_seller_listings, get_user_listing_status_counts, ) owner = db.first_or_404(db.select(User).where(User.public_id == public_id)) if current_user.is_authenticated and current_user.id == owner.id: return redirect(url_for("auth.profile")) owner_profile = owner.profile profile_data = { "full_name": owner_profile.full_name if owner_profile else owner.email.split("@")[0], "profile_picture": owner_profile.profile_picture if owner_profile else None, "bio": owner_profile.bio if owner_profile else "", "preferred_meeting_location": getattr(owner_profile, "preferred_meeting_location", "") if owner_profile else "", } counts = get_user_listing_status_counts(owner) listings = get_public_seller_listings(owner) from app.premium.services import get_premium_seller_ids from app.reviews.badges import get_badge_map, get_badges from app.reviews.reputation import get_recent_reviews_for_user, get_reputation reputation = get_reputation(owner.id) # cached; None → New Seller reputation_map = {owner.id: reputation} if reputation else {} return render_template( "auth/public_profile.html", owner=owner, profile=profile_data, active_count=counts["available"], sold_count=counts["sold"], total_count=counts["total"], listings=listings, premium_seller_ids=get_premium_seller_ids(listings), reputation_map=reputation_map, badge_map=get_badge_map(listings, reputation_map=reputation_map), reputation=reputation, badges=get_badges(owner), recent_reviews=get_recent_reviews_for_user(owner, limit=5), ) @auth_bp.route("/security", methods=["GET", "POST"]) @login_required def security(): """The account-security dashboard. Read/write logic lives in :mod:`app.auth.security_service` and the trusted- device service; this route only handles the change-password POST and gathers the dashboard context. Every panel is built on existing infrastructure (persisted security events, the OTP framework, trusted devices).""" password_form = PasswordChangeForm() if request.method == "POST" and "change_password" in request.form: if password_form.validate_on_submit(): if change_password( current_user, current_password=password_form.current_password.data, new_password=password_form.new_password.data, ): security_service.notify_password_changed(current_user) flash("Password updated successfully.", "success") else: flash("Current password is incorrect.", "danger") return redirect(url_for("auth.security")) return render_template( "auth/security.html", password_form=password_form, confirm_form=ConfirmPasswordForm(), action_form=TrustedDeviceActionForm(), rename_form=DeviceRenameForm(), two_factor_enabled=current_user.two_factor_enabled, security_score=security_service.compute_security_score(current_user), auth_timeline=security_service.build_auth_timeline(current_user), recent_activity=security_service.get_recent_activity(current_user, limit=10), recent_logins=security_service.get_login_history(current_user, page=1, per_page=6).items, failed_summary=security_service.get_failed_login_summary(current_user), active_sessions=security_service.get_active_sessions(current_user, request), devices=trusted_devices.list_active_devices(current_user), current_device_id=trusted_devices.current_device_id(current_user, request), notification_prefs=security_service.get_notification_prefs(current_user), notification_labels=security_service.NOTIFICATION_LABELS, password_policy_message=_PASSWORD_POLICY_MESSAGE, ) @auth_bp.route("/security/login-history", methods=["GET"]) @login_required def login_history(): """Full, paginated sign-in history (Login History feature).""" page = request.args.get("page", 1, type=int) pagination = security_service.get_login_history(current_user, page=page, per_page=20) return render_template( "auth/login_history.html", pagination=pagination, rows=pagination.items, method_label=security_service.method_label, ) @auth_bp.route("/security/report", methods=["GET"]) @login_required def security_report(): """Printable security report (browser 'Save as PDF'). Read-only summary.""" return render_template( "auth/security_report.html", user=current_user, generated_at=utcnow(), logins=security_service.get_login_history(current_user, page=1, per_page=30).items, devices=trusted_devices.list_active_devices(current_user), activity=security_service.get_recent_activity(current_user, limit=20), method_label=security_service.method_label, ) @auth_bp.route("/security/2fa/enable", methods=["POST"]) @login_required def two_factor_enable(): """Turn on Email-OTP two-factor. Reuses the login-OTP gate — once on, untrusted-device sign-ins require an emailed code.""" if not TrustedDeviceActionForm().validate_on_submit(): flash("Could not process that request. Please try again.", "danger") return redirect(url_for("auth.security")) if not current_user.two_factor_enabled: current_user.two_factor_enabled = True db.session.commit() log_auth_event(AuthEvent.TWO_FACTOR_ENABLED, user=current_user) security_service.notify_two_factor(current_user, enabled=True) flash("Two-factor authentication is now on. New devices will need an email code.", "success") return redirect(url_for("auth.security")) @auth_bp.route("/security/2fa/disable", methods=["POST"]) @login_required def two_factor_disable(): """Turn off two-factor — requires the account password (Security_Design).""" form = ConfirmPasswordForm() if not form.validate_on_submit(): flash("Please enter your password to disable two-factor.", "danger") return redirect(url_for("auth.security")) if not check_password_hash(current_user.password_hash, form.password.data): flash("Incorrect password. Two-factor authentication was not changed.", "danger") return redirect(url_for("auth.security")) if current_user.two_factor_enabled: current_user.two_factor_enabled = False db.session.commit() log_auth_event(AuthEvent.TWO_FACTOR_DISABLED, user=current_user) security_service.notify_two_factor(current_user, enabled=False) flash("Two-factor authentication has been turned off.", "info") return redirect(url_for("auth.security")) @auth_bp.route("/security/notifications", methods=["POST"]) @login_required def security_notifications(): """Save the user's security email-notification preferences (checkboxes).""" if not TrustedDeviceActionForm().validate_on_submit(): flash("Could not save your preferences. Please try again.", "danger") return redirect(url_for("auth.security")) prefs = {key: (key in request.form) for key in security_service.DEFAULT_NOTIFICATION_PREFS} security_service.set_notification_prefs(current_user, prefs) flash("Notification preferences saved.", "success") return redirect(url_for("auth.security")) @auth_bp.route("/security/sessions/sign-out-others", methods=["POST"]) @login_required def sign_out_other_sessions(): """Approximate 'sign out other sessions': revoke every trusted device except the current one, so other browsers must re-verify with an OTP.""" if not TrustedDeviceActionForm().validate_on_submit(): flash("Could not process that request. Please try again.", "danger") return redirect(url_for("auth.security")) current_id = trusted_devices.current_device_id(current_user, request) count = trusted_devices.revoke_all_for_user(current_user, except_device_id=current_id) if count: flash(f"Signed out {count} other device{'s' if count != 1 else ''}. " "They'll each need a verification code to sign in again.", "success") else: flash("There were no other devices to sign out.", "info") return redirect(url_for("auth.security")) @auth_bp.route("/security/devices", methods=["GET"]) @login_required def trusted_devices_page(): """Security → Trusted Devices: list this account's trusted browsers and let the user rename, remove one, remove others, or remove all. All logic lives in the trusted-device service; this route only renders and dispatches.""" return render_template( "auth/trusted_devices.html", devices=trusted_devices.list_active_devices(current_user), current_device_id=trusted_devices.current_device_id(current_user, request), action_form=TrustedDeviceActionForm(), rename_form=DeviceRenameForm(), ) @auth_bp.route("/security/devices//rename", methods=["POST"]) @login_required def trusted_device_rename(device_id: int): """Give a trusted device a friendly name.""" form = DeviceRenameForm() if not form.validate_on_submit(): flash("Please enter a valid device name (up to 80 characters).", "danger") return redirect(url_for("auth.trusted_devices_page")) if trusted_devices.rename_device(current_user, device_id, form.name.data): flash("Device renamed.", "success") else: flash("That device was not found.", "warning") return redirect(url_for("auth.trusted_devices_page")) @auth_bp.route("/security/devices/remove-others", methods=["POST"]) @login_required def trusted_device_remove_others(): """Remove every trusted device except the current one.""" if not TrustedDeviceActionForm().validate_on_submit(): flash("Could not process that request. Please try again.", "danger") return redirect(url_for("auth.trusted_devices_page")) current_id = trusted_devices.current_device_id(current_user, request) count = trusted_devices.revoke_all_for_user(current_user, except_device_id=current_id) if count: flash(f"Removed {count} other device{'s' if count != 1 else ''}. This device stays trusted.", "success") else: flash("There were no other devices to remove.", "info") return redirect(url_for("auth.trusted_devices_page")) @auth_bp.route("/security/devices//remove", methods=["POST"]) @login_required def trusted_device_remove(device_id: int): """Remove a single trusted device. Clears the browser cookie when the user removes the very device they're using.""" if not TrustedDeviceActionForm().validate_on_submit(): flash("Could not process that request. Please try again.", "danger") return redirect(url_for("auth.trusted_devices_page")) response = make_response(redirect(url_for("auth.trusted_devices_page"))) if trusted_devices.revoke_device(current_user, device_id, request, response): flash("Device removed. It will need a verification code next time it signs in.", "success") else: flash("That device was not found.", "warning") return response @auth_bp.route("/security/devices/remove-all", methods=["POST"]) @login_required def trusted_device_remove_all(): """Remove every trusted device on the account and clear this browser's cookie.""" if not TrustedDeviceActionForm().validate_on_submit(): flash("Could not process that request. Please try again.", "danger") return redirect(url_for("auth.trusted_devices_page")) count = trusted_devices.revoke_all_for_user(current_user) response = make_response(redirect(url_for("auth.trusted_devices_page"))) trusted_devices.clear_cookie(response) if count: flash(f"Removed {count} trusted device{'s' if count != 1 else ''}. " "New sign-ins will each require a verification code.", "success") else: flash("You have no trusted devices to remove.", "info") return response @auth_bp.route("/logout", methods=["POST"]) @login_required def logout(): logout_user() flash("You have been logged out.", "info") return redirect(url_for("auth.login"))