"""The signed-in user's own account (milestone 490): the time zone every page writes its times in, their email (milestone 495) and, for a local account, their password (milestone 497). Any role reaches it, for themselves only.""" from __future__ import annotations import time from datetime import datetime, timezone import bcrypt from quart import Blueprint, abort, current_app, redirect, render_template, request, session from steward.auth.middleware import require_role from steward.core import formatting from steward.models.users import User, UserRole account_bp = Blueprint("account", __name__, url_prefix="/account") _SOURCES = {"local": "a Steward password", "ldap": "LDAP", "oidc": "single sign-on"} def _password_matches(user: User, password: str) -> bool: try: return bcrypt.checkpw(password.encode(), user.password_hash.encode()) except ValueError: # not a bcrypt hash: no password matches it return False def _change_password(user: User, form) -> bool: """Apply the Password section: False when it was left blank, True when the password changed; ValueError says why it can't. Every other browser signed in as this account is signed out (#5427).""" from steward.auth.throttle import login_throttle from steward.core.proxy import client_ip from steward.settings.users import _hash, password_problem new = form.get("new_password") or "" if not new or user.auth_source != "local": return False ip = client_ip() if login_throttle.retry_after(ip, current_app.config.get("LOGIN_MAX_FAILURES", 5), current_app.config.get("LOGIN_LOCKOUT_MINUTES", 15) * 60): raise ValueError("Too many wrong passwords from this address. Try again later.") if not _password_matches(user, form.get("current_password") or ""): login_throttle.failed(ip) raise ValueError("Your current password isn't right, so nothing was saved.") problem = password_problem(new, form.get("new_confirm") or "") if problem: raise ValueError(problem) user.password_hash = _hash(new) user.sessions_since = datetime.now(timezone.utc) return True async def _me(db) -> User: # A share link opens widgets read-only; it has no account. user_id = session.get("user_id") if not user_id: abort(403) user = await db.get(User, user_id) if user is None: abort(403) return user @account_bp.get("/") @require_role(UserRole.viewer) async def account(error: str | None = None, status: int = 200): from steward.core.mail import role_defaults from steward.core.settings import get_all_settings async with current_app.db_sessionmaker() as db: user = await _me(db) defaults = role_defaults(user, await get_all_settings(db)) tz = formatting.zone(user.time_zone) zone_name = user.time_zone or "UTC" return await render_template( "account/index.html", user=user, source=_SOURCES.get(user.auth_source, user.auth_source), zone_name=zone_name, zone_title=zone_name.split("/", 1)[-1].replace("_", " "), zone_label=formatting.tz_label(tz), zones=formatting.zones_by_region(), defaults=defaults, error=error, min_password=_min_password(), done=_DONE.get(request.args.get("done", ""))), status _DONE = {"saved": "Saved.", "password": "Saved. Your password is changed; anywhere else you were signed in, you're signed out."} def _min_password() -> int: from steward.settings.users import MIN_PASSWORD return MIN_PASSWORD @account_bp.post("/") @require_role(UserRole.viewer) async def account_save(): """The page's one form: your time zone, your address, whether you're emailed alerts and the weekly report (milestone 495), and a new password (milestone 497). Nothing is saved unless all of it is valid.""" from sqlalchemy.exc import IntegrityError from steward.core.audit import log_audit from steward.core.mail import apply_email_form form = await request.form name = (form.get("time_zone") or "").strip() if name not in formatting.zone_names(): return await account(f"{name or 'That'} is not a time zone Steward knows.", 400) try: async with current_app.db_sessionmaker() as db: async with db.begin(): user = await _me(db) changes, error = apply_email_form(user, form) if error: raise ValueError(error) before = user.time_zone user.time_zone = name password_changed = _change_password(user, form) except ValueError as exc: return await account(str(exc), 400) except IntegrityError: return await account("Another account already uses that address.", 400) session["tz"] = name if password_changed: session["login_at"] = time.time() # this session carries on under the new password await log_audit(current_app, user.id, user.username, "account.password_changed") if before != name: await log_audit(current_app, user.id, user.username, "account.time_zone", detail={"time_zone": name, "was": before or "UTC"}) if changes: await log_audit(current_app, user.id, user.username, "account.email", detail=changes) return redirect("/account/?done=" + ("password" if password_changed else "saved")) @account_bp.post("/time-zone/adopt") @require_role(UserRole.viewer) async def adopt_time_zone(): """The browser's zone, saved for an account that has none yet: base.html sends it once, on the first page after a sign-in without one. A zone set on the Account page is never replaced. Answers whether it was saved.""" name = ((await request.get_json(silent=True)) or {}).get("time_zone") or "" if name not in formatting.zone_names(): return {"adopted": False}, 400 async with current_app.db_sessionmaker() as db: async with db.begin(): user = await _me(db) if user.time_zone is not None: return {"adopted": False} user.time_zone = name session["tz"] = name return {"adopted": True}