bvandeusenandClaude Opus 5.5
beae8a33f3
Operator review of the Account page: two sections with two different save buttons, one stuck to the next card and one beside a squeezed hint. - Account: time zone and Email are one form; POST /account/ saves both, nothing unless all of it is valid. /account/email is gone. - Users › account: Name and Email are one form; the rename route also saves the email switches. /settings/users/<id>/email is gone. - The Email section is a plain section now (email_section); its hint sits under the switches, names the zone as the picker does, and links Notifications only for an admin. - apply_email_form leaves a switch the form doesn't carry as it is. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
307 lines
14 KiB
Python
307 lines
14 KiB
Python
"""Settings › Users (#5336): who has an account, and an admin's changes to one.
|
||
|
||
A local account's role, name and password are set here. An LDAP or single
|
||
sign-on account's role comes from its provider's group mapping at every
|
||
sign-in, so it is shown, not edited. Nobody can disable, demote or delete
|
||
their own account, and the last active admin cannot be demoted or disabled,
|
||
so Steward always keeps someone who can get back in. Every change is audited.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from datetime import datetime, timezone
|
||
|
||
import bcrypt
|
||
from quart import Blueprint, abort, current_app, redirect, render_template, request, session
|
||
from sqlalchemy import func, or_, select
|
||
from sqlalchemy.exc import IntegrityError
|
||
|
||
from steward.auth.middleware import require_role
|
||
from steward.core import mail
|
||
from steward.core.audit import audit_request
|
||
from steward.core.settings import get_all_settings
|
||
from steward.models.alerts import AlertRule, MaintenanceWindow
|
||
from steward.models.ansible import AnsibleRun
|
||
from steward.models.api_tokens import ApiToken
|
||
from steward.models.audit import AuditEvent
|
||
from steward.models.users import User, UserRole
|
||
|
||
users_bp = Blueprint("users", __name__, url_prefix="/settings/users")
|
||
|
||
SOURCES = {"local": "Steward password", "ldap": "LDAP", "oidc": "Single sign-on"}
|
||
MIN_PASSWORD = 8
|
||
|
||
|
||
def refusal(change: str, target: User, actor_id: str | None, active_admins: int) -> str | None:
|
||
"""Why `change` ("demote", "disable" or "delete") may not be made to
|
||
`target` by the signed-in admin, or None when it may."""
|
||
if target.id == actor_id:
|
||
return {"demote": "You can't lower your own role. Another admin can.",
|
||
"disable": "You can't disable your own account.",
|
||
"delete": "You can't delete your own account."}[change]
|
||
if target.role == UserRole.admin and target.is_active and active_admins <= 1:
|
||
return f"{target.username} is the only active admin. Make someone else an admin first."
|
||
return None
|
||
|
||
|
||
def password_problem(password: str, confirm: str) -> str | None:
|
||
if len(password) < MIN_PASSWORD:
|
||
return f"A password needs at least {MIN_PASSWORD} characters."
|
||
if password != confirm:
|
||
return "The two passwords don't match."
|
||
return None
|
||
|
||
|
||
def _hash(password: str) -> str:
|
||
return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
|
||
|
||
|
||
async def _active_admins(db) -> int:
|
||
return (await db.execute(select(func.count()).select_from(User).where(
|
||
User.role == UserRole.admin, User.is_active.is_(True)))).scalar_one()
|
||
|
||
|
||
async def _last_sign_ins(db) -> dict[str, datetime]:
|
||
"""Each user's newest sign-in, from the audit log (kept as long as it is)."""
|
||
rows = await db.execute(select(AuditEvent.user_id, func.max(AuditEvent.timestamp))
|
||
.where(AuditEvent.action == "auth.login", AuditEvent.user_id.isnot(None))
|
||
.group_by(AuditEvent.user_id))
|
||
return dict(rows.all())
|
||
|
||
|
||
async def _token_counts(db) -> dict[str, int]:
|
||
"""Each user's AI access tokens that still work."""
|
||
now = datetime.now(timezone.utc)
|
||
rows = await db.execute(select(ApiToken.user_id, func.count()).where(
|
||
ApiToken.revoked_at.is_(None),
|
||
or_(ApiToken.expires_at.is_(None), ApiToken.expires_at > now)).group_by(ApiToken.user_id))
|
||
return dict(rows.all())
|
||
|
||
|
||
async def held_by(db, user_id: str) -> list[str]:
|
||
"""What keeps an account from being deleted: records that name who made
|
||
them and must keep naming them, in words ("3 alert rules")."""
|
||
counts = [
|
||
(await db.execute(select(func.count()).where(AlertRule.created_by == user_id))).scalar_one(),
|
||
(await db.execute(select(func.count()).where(MaintenanceWindow.created_by == user_id))).scalar_one(),
|
||
(await db.execute(select(func.count()).where(AnsibleRun.triggered_by == user_id))).scalar_one(),
|
||
]
|
||
words = (("alert rule", "alert rules"), ("maintenance window", "maintenance windows"),
|
||
("playbook run", "playbook runs"))
|
||
return [f"{n} {one if n == 1 else many}" for n, (one, many) in zip(counts, words) if n]
|
||
|
||
|
||
async def _target(db, user_id: str) -> User:
|
||
user = await db.get(User, user_id)
|
||
if user is None:
|
||
abort(404)
|
||
return user
|
||
|
||
|
||
@users_bp.get("/")
|
||
@require_role(UserRole.admin)
|
||
async def users_list():
|
||
async with current_app.db_sessionmaker() as db:
|
||
users = (await db.execute(select(User))).scalars().all()
|
||
last = await _last_sign_ins(db)
|
||
tokens = await _token_counts(db)
|
||
settings = await get_all_settings(db)
|
||
users = sorted(users, key=lambda u: (not u.is_active, u.username.lower()))
|
||
gets = {u.id: [label for kind, label in (("alerts", "alerts"), ("report", "report"))
|
||
if u.is_active and mail.wants(u, kind, settings)] for u in users}
|
||
return await render_template(
|
||
"settings/users.html", users=users, last=last, tokens=tokens, sources=SOURCES,
|
||
gets=gets, reachable={u.id for u in users if mail.is_real_address(u.email)},
|
||
admins=sum(1 for u in users if u.role == UserRole.admin and u.is_active),
|
||
disabled=sum(1 for u in users if not u.is_active))
|
||
|
||
|
||
async def _new_page(error: str | None = None, form=None, status: int = 200):
|
||
return await render_template("settings/user_new.html", error=error, form=form or {},
|
||
roles=list(UserRole), min_password=MIN_PASSWORD), status
|
||
|
||
|
||
@users_bp.get("/new")
|
||
@require_role(UserRole.admin)
|
||
async def user_new():
|
||
return await _new_page()
|
||
|
||
|
||
@users_bp.post("/new")
|
||
@require_role(UserRole.admin)
|
||
async def user_create():
|
||
form = await request.form
|
||
username = (form.get("username") or "").strip()
|
||
email = (form.get("email") or "").strip()
|
||
try:
|
||
role = UserRole(form.get("role") or "viewer")
|
||
except ValueError:
|
||
return await _new_page("Pick a role.", form, 400)
|
||
if not username or not email:
|
||
return await _new_page("A username and an email are both needed.", form, 400)
|
||
if len(username) > 64 or len(email) > 255:
|
||
return await _new_page("That username or email is too long.", form, 400)
|
||
problem = password_problem(form.get("password") or "", form.get("confirm") or "")
|
||
if problem:
|
||
return await _new_page(problem, form, 400)
|
||
user = User(username=username, email=email, role=role, auth_source="local",
|
||
password_hash=_hash(form.get("password")))
|
||
try:
|
||
async with current_app.db_sessionmaker() as db:
|
||
async with db.begin():
|
||
db.add(user)
|
||
except IntegrityError:
|
||
return await _new_page(f"An account already uses the username {username} or that email.", form, 400)
|
||
await audit_request("user.created", "user", user.id, {"username": username, "role": role.value})
|
||
return redirect(f"/settings/users/{user.id}?done=created")
|
||
|
||
|
||
_DONE = {"created": "Account added.", "saved": "Saved.", "role": "Role changed.",
|
||
"password": "Password set. It works from their next sign-in.",
|
||
"disabled": "Account disabled. They are signed out at their next click.",
|
||
"enabled": "Account enabled."}
|
||
|
||
|
||
async def _user_page(user_id: str, error: str | None = None, status: int = 200):
|
||
async with current_app.db_sessionmaker() as db:
|
||
user = await _target(db, user_id)
|
||
admins = await _active_admins(db)
|
||
last = (await _last_sign_ins(db)).get(user.id)
|
||
tokens = (await _token_counts(db)).get(user.id, 0)
|
||
holds = await held_by(db, user.id)
|
||
defaults = mail.role_defaults(user, await get_all_settings(db))
|
||
me = session.get("user_id")
|
||
return await render_template(
|
||
"settings/user.html", user=user, last_sign_in=last, tokens=tokens, holds=holds,
|
||
defaults=defaults, reachable=mail.is_real_address(user.email),
|
||
source=SOURCES.get(user.auth_source, user.auth_source), roles=list(UserRole),
|
||
min_password=MIN_PASSWORD, error=error, done=_DONE.get(request.args.get("done", "")),
|
||
demote_refused=refusal("demote", user, me, admins),
|
||
disable_refused=refusal("disable", user, me, admins) if user.is_active else None,
|
||
delete_refused=refusal("delete", user, me, admins)), status
|
||
|
||
|
||
@users_bp.get("/<user_id>")
|
||
@require_role(UserRole.admin)
|
||
async def user_detail(user_id: str):
|
||
return await _user_page(user_id)
|
||
|
||
|
||
@users_bp.post("/<user_id>")
|
||
@require_role(UserRole.admin)
|
||
async def user_save(user_id: str):
|
||
"""The page's one form, for any account: its username and email, and
|
||
whether it is emailed alerts and the weekly report (milestone 495). The
|
||
address may be a provider's placeholder, so it isn't held to mail's test
|
||
here; the account just gets no email until it is real."""
|
||
form = await request.form
|
||
username = (form.get("username") or "").strip()
|
||
email = (form.get("email") or "").strip()
|
||
if not username or not email or len(username) > 64 or len(email) > 255:
|
||
return await _user_page(user_id, "A username (up to 64 characters) and an email are both needed.", 400)
|
||
switches = {k: v for k, v in form.items() if k != "email"}
|
||
try:
|
||
async with current_app.db_sessionmaker() as db:
|
||
async with db.begin():
|
||
user = await _target(db, user_id)
|
||
emailed, error = mail.apply_email_form(user, switches)
|
||
if error:
|
||
raise ValueError(error)
|
||
before = {"username": user.username, "email": user.email}
|
||
user.username, user.email = username, email
|
||
except ValueError as exc:
|
||
return await _user_page(user_id, str(exc), 400)
|
||
except IntegrityError:
|
||
return await _user_page(user_id, f"Another account already uses the username {username} or that email.", 400)
|
||
after = {"username": username, "email": email}
|
||
if after != before:
|
||
await audit_request("user.renamed", "user", user_id,
|
||
{k: v for k, v in after.items() if before[k] != v}
|
||
| {f"old_{k}": before[k] for k in after if before[k] != after[k]})
|
||
if user_id == session.get("user_id"):
|
||
session["username"] = username
|
||
if emailed:
|
||
await audit_request("user.email_settings", "user", user_id, {"username": username} | emailed)
|
||
return redirect(f"/settings/users/{user_id}?done=saved")
|
||
|
||
|
||
@users_bp.post("/<user_id>/role")
|
||
@require_role(UserRole.admin)
|
||
async def user_role(user_id: str):
|
||
try:
|
||
role = UserRole((await request.form).get("role") or "")
|
||
except ValueError:
|
||
return await _user_page(user_id, "Pick a role.", 400)
|
||
async with current_app.db_sessionmaker() as db:
|
||
async with db.begin():
|
||
user = await _target(db, user_id)
|
||
if user.auth_source != "local":
|
||
error = f"{SOURCES[user.auth_source]} sets this account's role at each sign-in."
|
||
elif role != UserRole.admin and user.role == UserRole.admin:
|
||
error = refusal("demote", user, session.get("user_id"), await _active_admins(db))
|
||
else:
|
||
error = None
|
||
if error is None:
|
||
before, user.role = user.role, role
|
||
if error:
|
||
return await _user_page(user_id, error, 400)
|
||
if before != role:
|
||
await audit_request("user.role_changed", "user", user_id,
|
||
{"username": user.username, "from": before.value, "to": role.value})
|
||
return redirect(f"/settings/users/{user_id}?done=role")
|
||
|
||
|
||
@users_bp.post("/<user_id>/password")
|
||
@require_role(UserRole.admin)
|
||
async def user_password(user_id: str):
|
||
form = await request.form
|
||
problem = password_problem(form.get("password") or "", form.get("confirm") or "")
|
||
async with current_app.db_sessionmaker() as db:
|
||
async with db.begin():
|
||
user = await _target(db, user_id)
|
||
if user.auth_source != "local":
|
||
problem = f"This account signs in with {SOURCES[user.auth_source]}; it has no Steward password."
|
||
if problem is None:
|
||
user.password_hash = _hash(form.get("password"))
|
||
if problem:
|
||
return await _user_page(user_id, problem, 400)
|
||
await audit_request("user.password_reset", "user", user_id, {"username": user.username})
|
||
return redirect(f"/settings/users/{user_id}?done=password")
|
||
|
||
|
||
@users_bp.post("/<user_id>/active")
|
||
@require_role(UserRole.admin)
|
||
async def user_active(user_id: str):
|
||
"""Disable an active account, or enable a disabled one. A disabled user is
|
||
signed out at their next request and their AI tokens stop working."""
|
||
async with current_app.db_sessionmaker() as db:
|
||
async with db.begin():
|
||
user = await _target(db, user_id)
|
||
error = (refusal("disable", user, session.get("user_id"), await _active_admins(db))
|
||
if user.is_active else None)
|
||
if error is None:
|
||
user.is_active = not user.is_active
|
||
if error:
|
||
return await _user_page(user_id, error, 400)
|
||
action = "enabled" if user.is_active else "disabled"
|
||
await audit_request(f"user.{action}", "user", user_id, {"username": user.username})
|
||
return redirect(f"/settings/users/{user_id}?done={action}")
|
||
|
||
|
||
@users_bp.post("/<user_id>/delete")
|
||
@require_role(UserRole.admin)
|
||
async def user_delete(user_id: str):
|
||
async with current_app.db_sessionmaker() as db:
|
||
async with db.begin():
|
||
user = await _target(db, user_id)
|
||
error = refusal("delete", user, session.get("user_id"), await _active_admins(db))
|
||
holds = await held_by(db, user_id)
|
||
if error is None and holds:
|
||
error = (f"{user.username} made {', '.join(holds)}, and Steward keeps who did. "
|
||
"Disable the account instead.")
|
||
if error is None:
|
||
await db.delete(user)
|
||
if error:
|
||
return await _user_page(user_id, error, 400)
|
||
await audit_request("user.deleted", "user", user_id, {"username": user.username})
|
||
return redirect("/settings/users/")
|