FabledSteward/steward/account/routes.py

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