diff --git a/steward/app.py b/steward/app.py index 60e6593..93edcfd 100644 --- a/steward/app.py +++ b/steward/app.py @@ -215,6 +215,7 @@ def create_app( from .ansible.routes import ansible_bp from .ansible.inventory_routes import inventory_bp from .settings.routes import settings_bp + from .settings.users import users_bp from .audit.routes import audit_bp from .ai_access.routes import ai_access_bp from .account.routes import account_bp @@ -228,6 +229,7 @@ def create_app( app.register_blueprint(ansible_bp) app.register_blueprint(inventory_bp) app.register_blueprint(settings_bp) + app.register_blueprint(users_bp) app.register_blueprint(audit_bp) app.register_blueprint(ai_access_bp) app.register_blueprint(account_bp) diff --git a/steward/audit/service.py b/steward/audit/service.py index 3cf6175..8f2cbe6 100644 --- a/steward/audit/service.py +++ b/steward/audit/service.py @@ -19,6 +19,7 @@ from steward.models.ansible_schedule import AnsibleSchedule from steward.models.api_tokens import ApiToken from steward.models.hosts import Host from steward.models.monitors import Monitor +from steward.models.users import User logger = logging.getLogger(__name__) @@ -33,6 +34,7 @@ _NAMED = { "ansible_group": (AnsibleGroup, "/ansible/inventory/groups/{}"), "ansible_schedule": (AnsibleSchedule, "/ansible/schedules"), "api_token": (ApiToken, "/ai-access/"), + "user": (User, "/settings/users/{}"), } _SELF_NAMED = {"plugin": "/settings/plugins/", "ansible_playbook": None} @@ -80,8 +82,9 @@ async def entity_labels(db, items: list[dict]) -> dict[tuple[str, str], dict]: keys = [int(i) for i in wanted if i.isdigit()] if etype == "api_token" else list(wanted) if not keys: continue + name_col = model.username if model is User else model.name for row_id, name in (await db.execute( - select(key_col, model.name).where(key_col.in_(keys)))).all(): + select(key_col, name_col).where(key_col.in_(keys)))).all(): labels[(etype, str(row_id))] = {"label": name, "url": page.format(row_id), "gone": False} elif etype == "ansible_run": runs = (await db.execute( @@ -116,6 +119,13 @@ ACTIONS = { "auth.login_failed": "failed to sign in", "auth.setup": "set up Steward", "account.time_zone": "set their time zone", + "user.created": "added user", + "user.renamed": "renamed user", + "user.role_changed": "changed the role of user", + "user.password_reset": "set a new password for user", + "user.disabled": "disabled user", + "user.enabled": "enabled user", + "user.deleted": "deleted user", "host.created": "added host", "host.renamed": "renamed host", "host.deleted": "deleted host", @@ -170,7 +180,7 @@ ACTIONS = { } # The filter's choices: an area, matched as a prefix of the action code. -AREAS = (("auth.", "Sign-ins"), ("account.", "Accounts"), ("host.", "Hosts"), ("monitor.", "Monitors"), +AREAS = (("auth.", "Sign-ins"), ("account.", "Accounts"), ("user.", "Users"), ("host.", "Hosts"), ("monitor.", "Monitors"), ("alert_rule.", "Alert rules"), ("maintenance_window.", "Maintenance"), ("ansible.run", "Playbook runs"), ("ansible.", "Ansible (all)"), ("settings.", "Settings"), ("plugin.", "Plugins"), ("api_token.", "AI access"), diff --git a/steward/settings/users.py b/steward/settings/users.py new file mode 100644 index 0000000..34be7d5 --- /dev/null +++ b/steward/settings/users.py @@ -0,0 +1,287 @@ +"""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.audit import audit_request +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) + users = sorted(users, key=lambda u: (not u.is_active, u.username.lower())) + return await render_template( + "settings/users.html", users=users, last=last, tokens=tokens, sources=SOURCES, + 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) + me = session.get("user_id") + return await render_template( + "settings/user.html", user=user, last_sign_in=last, tokens=tokens, holds=holds, + 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("/") +@require_role(UserRole.admin) +async def user_detail(user_id: str): + return await _user_page(user_id) + + +@users_bp.post("/") +@require_role(UserRole.admin) +async def user_save(user_id: str): + """Rename: the username and email, for any account.""" + 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) + try: + async with current_app.db_sessionmaker() as db: + async with db.begin(): + user = await _target(db, user_id) + before = {"username": user.username, "email": user.email} + user.username, user.email = username, email + 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 + return redirect(f"/settings/users/{user_id}?done=saved") + + +@users_bp.post("//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("//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("//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("//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/") diff --git a/steward/templates/settings/_tabs.html b/steward/templates/settings/_tabs.html index af050a4..f12a57b 100644 --- a/steward/templates/settings/_tabs.html +++ b/steward/templates/settings/_tabs.html @@ -8,6 +8,7 @@ ("notifications", "Notifications", "/settings/notifications/"), ("reports", "Reports", "/settings/reports/"), ("auth", "Authentication", "/settings/auth/"), + ("users", "Users", "/settings/users/"), ("ansible", "Ansible", "/settings/ansible/"), ("plugins", "Plugins", "/settings/plugins/")], active, "Settings") }} {%- endmacro %} diff --git a/steward/templates/settings/user.html b/steward/templates/settings/user.html new file mode 100644 index 0000000..38539b4 --- /dev/null +++ b/steward/templates/settings/user.html @@ -0,0 +1,116 @@ +{# settings/user.html — one account (#5336): who it is and how it signs in, + then its name, role and password, disabling it, and Delete at the foot. + A refusal (yourself, the last admin, records that name it) is said beside + the control rather than discovered by pressing it. #} +{% extends "base.html" %} +{% from "_macros.html" import crumbs, page_header, status_badge %} +{% from "settings/_tabs.html" import settings_tabs %} +{% block title %}{{ user.username }} — Users — Steward{% endblock %} +{% block section_tabs %}{{ settings_tabs("users") }}{% endblock %} +{% block breadcrumb %}{{ crumbs([("Settings", "/settings/"), ("Users", "/settings/users/"), (user.username, "")]) }}{% endblock %} +{% block content %} +{% set local = user.auth_source == "local" %} +{{ page_header(user.username) }} + +

+ {% if user.is_active %}{{ user.role.value|capitalize }}{% else %}{{ status_badge("inactive", "Disabled") }} {{ user.role.value }}{% endif %} + · signs in with {{ source }} + · {% if last_sign_in %}last signed in {{ last_sign_in|ago }}{% else %}no sign-in on record{% endif %} + · {% if not tokens %}no AI tokens{% elif user.id == session.user_id %}{{ tokens }} AI token{{ 's' if tokens != 1 }}{% else %}{{ tokens }} AI token{{ 's' if tokens != 1 }}{% endif %} + · added {{ user.created_at|when }} +

+ +{% if error %} +{% elif done %}
{{ done }}
{% endif %} + +
{{ csrf_field() }} +
+

Name

+
+
+ + +
+
+ + +
+
+ {% if not local %}

{{ source }} finds this account by its own id, not by name, so renaming it here does not stop it signing in.

{% endif %} + +
+
+ +
+

Role

+ {% if not local %} +

{{ user.role.value|capitalize }}. {{ source }} sets it from the account's groups at every sign-in; the mapping is in Settings › Authentication.

+ {% else %} +
{{ csrf_field() }} + + +
+ {% if demote_refused %}

{{ demote_refused }}

{% endif %} + {% endif %} +
+ +{% if local %} +
{{ csrf_field() }} +
+

Password

+
+
+ + + At least {{ min_password }} characters. Give it to them yourself; Steward sends no email for it. +
+
+ + +
+
+ +
+
+{% endif %} + +
+

{{ "Disable" if user.is_active else "Enable" }}

+ {% if user.is_active %} +

A disabled account is signed out at its next click and its AI tokens stop working. Everything it made stays.

+ {% if disable_refused %} +

{{ disable_refused }}

+ {% else %} +
{{ csrf_field() }} + +
+ {% endif %} + {% else %} +

This account can't sign in, and its AI tokens don't work.

+
{{ csrf_field() }} + +
+ {% endif %} +
+ +
+

Delete this account

+ {% if delete_refused %} +

{{ delete_refused }}

+ {% elif holds %} +

{{ user.username }} made {{ holds|join(", ") }}, and Steward keeps who did, so this account can't be deleted. Disable it instead.

+ {% else %} +

Its AI tokens go with it. The audit log keeps its name.

+
{{ csrf_field() }} + +
+ {% endif %} +
+{% endblock %} diff --git a/steward/templates/settings/user_new.html b/steward/templates/settings/user_new.html new file mode 100644 index 0000000..40d915e --- /dev/null +++ b/steward/templates/settings/user_new.html @@ -0,0 +1,54 @@ +{# settings/user_new.html — add an account that signs in with a Steward password (#5336). #} +{% extends "base.html" %} +{% from "_macros.html" import crumbs, page_header %} +{% from "settings/_tabs.html" import settings_tabs %} +{% block title %}Add user — Settings — Steward{% endblock %} +{% block section_tabs %}{{ settings_tabs("users") }}{% endblock %} +{% block breadcrumb %}{{ crumbs([("Settings", "/settings/"), ("Users", "/settings/users/"), ("Add user", "")]) }}{% endblock %} +{% block content %} +{{ page_header("Add user", "Signs in with a Steward password.") }} +{% if error %}{% endif %} + +
{{ csrf_field() }} +
+

Account

+
+
+ + +
+
+ + +
+
+
+ + + A viewer reads; an operator also runs playbooks and changes monitors, hosts and alerts; an admin also changes Settings and users. +
+
+
+

Password

+
+
+ + + At least {{ min_password }} characters. Give it to them yourself; Steward sends no email for it. +
+
+ + +
+
+
+
+ + Cancel +
+
+{% endblock %} diff --git a/steward/templates/settings/users.html b/steward/templates/settings/users.html new file mode 100644 index 0000000..a7a7f18 --- /dev/null +++ b/steward/templates/settings/users.html @@ -0,0 +1,39 @@ +{# settings/users.html — Settings › Users (#5336): every account, active ones + first. A row opens the account's page. #} +{% extends "base.html" %} +{% from "_macros.html" import crumbs, page_header, status_badge %} +{% from "settings/_tabs.html" import settings_tabs %} +{% block title %}Settings — Users — Steward{% endblock %} +{% block section_tabs %}{{ settings_tabs("users") }}{% endblock %} +{% block breadcrumb %}{{ crumbs([("Settings", "/settings/"), ("Users", "")]) }}{% endblock %} +{% block content %} +{% call page_header("Users") %}Add user{% endcall %} + +

+ {{ users|length }} account{{ 's' if users|length != 1 }} + · {{ admins }} active admin{{ 's' if admins != 1 }} + {%- if disabled %} · {{ disabled }} disabled{% endif %} +

+ +
+ + + + {% for u in users %} + + + + + + + + {% endfor %} + +
UserSigns in withRoleLast sign-inAI tokens
+ {{ u.username }} + {%- if u.id == session.user_id %} (you){% endif %} +
{{ u.email }}
+ 		{{ sources.get(u.auth_source, u.auth_source) }}{{ u.role.value|capitalize }}{% if not u.is_active %} {{ status_badge("inactive", "Disabled") }}{% endif %}{% if last.get(u.id) %}{{ last[u.id]|ago }}{% else %}—{% endif %}{{ tokens.get(u.id, 0) or "—" }}
+
+

Accounts that sign in with LDAP or single sign-on are added at their first sign-in, with the role their provider's groups give them (Settings › Authentication).

+{% endblock %} diff --git a/tests/core/test_user_guards.py b/tests/core/test_user_guards.py new file mode 100644 index 0000000..35634af --- /dev/null +++ b/tests/core/test_user_guards.py @@ -0,0 +1,29 @@ +"""Settings › Users keeps someone able to get back in (#5336): nobody can +disable, demote or delete their own account, and the last active admin +cannot be demoted or disabled.""" +from steward.models.users import User, UserRole +from steward.settings.users import password_problem, refusal + + +def _u(uid, role=UserRole.admin, active=True): + return User(id=uid, username=uid, email=f"{uid}@t", password_hash="x", role=role, is_active=active) + + +def test_nobody_changes_their_own_access(): + me = _u("me") + for change in ("demote", "disable", "delete"): + assert "own" in refusal(change, me, "me", active_admins=5) + + +def test_the_last_active_admin_stays(): + only = _u("only") + assert "only active admin" in refusal("demote", only, "someone-else", active_admins=1) + assert refusal("demote", only, "someone-else", active_admins=2) is None + assert refusal("disable", _u("v", role=UserRole.viewer), "me", active_admins=1) is None + assert refusal("delete", _u("gone", active=False), "me", active_admins=1) is None + + +def test_a_password_is_long_enough_and_typed_twice(): + assert "at least" in password_problem("short", "short") + assert "match" in password_problem("long enough", "long enougj") + assert password_problem("long enough", "long enough") is None diff --git a/tests/integration/test_users_page.py b/tests/integration/test_users_page.py new file mode 100644 index 0000000..e38bc6d --- /dev/null +++ b/tests/integration/test_users_page.py @@ -0,0 +1,170 @@ +"""Integration: Settings › Users (#5336) against real Postgres. An admin adds +a local account, changes its role, sets its password, disables and deletes +it; their own account and an account that made records are refused, and +every change is audited.""" +from __future__ import annotations + +import asyncio +import json +import os +import time +import uuid +from datetime import datetime, timedelta, timezone + +import bcrypt +import pytest + +pytestmark = pytest.mark.integration +CSRF = {"X-CSRF-Token": "test-token"} + + +@pytest.fixture +def app(): + if not os.environ.get("STEWARD_DATABASE_URL"): + pytest.skip("needs Postgres") + from steward.app import create_app + return create_app(testing=False) + + +async def _client(app, role="admin"): + from steward.models.users import User, UserRole + uid = str(uuid.uuid4()) + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add(User(id=uid, username=f"adm-{uid[:8]}", email=f"{uid}@test", + password_hash="x", role=UserRole(role), is_active=True)) + client = app.test_client() + async with client.session_transaction() as sess: + sess.update(user_id=uid, user_role=role, username=f"adm-{uid[:8]}", + _csrf="test-token", login_at=time.time()) + return client, uid + + +async def _user(app, user_id): + from steward.models.users import User + async with app.db_sessionmaker() as s: + return await s.get(User, user_id) + + +async def _audits(app, uid): + from sqlalchemy import select + from steward.models.audit import AuditEvent + async with app.db_sessionmaker() as s: + rows = (await s.execute(select(AuditEvent).where(AuditEvent.user_id == uid) + .order_by(AuditEvent.timestamp))).scalars().all() + return [(r.action, json.loads(r.detail_json or "{}")) for r in rows] + + +def test_an_admin_adds_changes_and_deletes_a_local_account(app): + async def _go(): + client, me = await _client(app) + name = f"new-{uuid.uuid4().hex[:8]}" + resp = await client.post("/settings/users/new", headers=CSRF, form={ + "username": name, "email": f"{name}@test", "role": "viewer", + "password": "short", "confirm": "short"}) + assert resp.status_code == 400 and "at least 8 characters" in await resp.get_data(as_text=True) + + resp = await client.post("/settings/users/new", headers=CSRF, form={ + "username": name, "email": f"{name}@test", "role": "viewer", + "password": "correct horse", "confirm": "correct horse"}) + assert resp.status_code == 302 + uid = resp.headers["Location"].split("/settings/users/")[1].split("?")[0] + user = await _user(app, uid) + assert (user.username, user.role.value, user.auth_source) == (name, "viewer", "local") + assert bcrypt.checkpw(b"correct horse", user.password_hash.encode()) + + page = await (await client.get("/settings/users/")).get_data(as_text=True) + assert f'href="/settings/users/{uid}"' in page and 'aria-label="Settings"' in page + + assert (await client.post(f"/settings/users/{uid}/role", headers=CSRF, + form={"role": "operator"})).status_code == 302 + assert (await _user(app, uid)).role.value == "operator" + + resp = await client.post(f"/settings/users/{uid}/password", headers=CSRF, + form={"password": "battery staple", "confirm": "battery stapler"}) + assert resp.status_code == 400 and "match" in await resp.get_data(as_text=True) + assert (await client.post(f"/settings/users/{uid}/password", headers=CSRF, + form={"password": "battery staple", "confirm": "battery staple"})).status_code == 302 + assert bcrypt.checkpw(b"battery staple", (await _user(app, uid)).password_hash.encode()) + + assert (await client.post(f"/settings/users/{uid}/active", headers=CSRF)).status_code == 302 + assert not (await _user(app, uid)).is_active + assert (await client.post(f"/settings/users/{uid}/active", headers=CSRF)).status_code == 302 + assert (await _user(app, uid)).is_active + + assert (await client.post(f"/settings/users/{uid}", headers=CSRF, + form={"username": name + "-x", "email": f"{name}@test"})).status_code == 302 + assert (await _user(app, uid)).username == name + "-x" + + assert (await client.post(f"/settings/users/{uid}/delete", headers=CSRF)).status_code == 302 + assert await _user(app, uid) is None + + assert [a for a, _ in await _audits(app, me)] == [ + "user.created", "user.role_changed", "user.password_reset", "user.disabled", + "user.enabled", "user.renamed", "user.deleted"] + + asyncio.run(_go()) + + +def test_an_admin_cannot_disable_demote_or_delete_themselves(app): + async def _go(): + client, me = await _client(app) + for path, form, says in ((f"/settings/users/{me}/active", {}, "your own account"), + (f"/settings/users/{me}/role", {"role": "viewer"}, "your own role"), + (f"/settings/users/{me}/delete", {}, "your own account")): + resp = await client.post(path, headers=CSRF, form=form) + assert resp.status_code == 400 and says in await resp.get_data(as_text=True), path + me_now = await _user(app, me) + assert me_now.is_active and me_now.role.value == "admin" + + asyncio.run(_go()) + + +def test_an_account_that_made_records_is_disabled_not_deleted(app): + from steward.models.alerts import MaintenanceWindow + + async def _go(): + client, _ = await _client(app) + _, other = await _client(app, role="operator") + now = datetime.now(timezone.utc) + async with app.db_sessionmaker() as s: + async with s.begin(): + # Over already, so it silences no other test's alerts. + s.add(MaintenanceWindow(name="patching", start_at=now - timedelta(days=2), + end_at=now - timedelta(days=1), created_by=other)) + page = await (await client.get(f"/settings/users/{other}")).get_data(as_text=True) + assert "1 maintenance window" in page and "Delete account" not in page + resp = await client.post(f"/settings/users/{other}/delete", headers=CSRF) + assert resp.status_code == 400 and "Disable the account instead" in await resp.get_data(as_text=True) + assert await _user(app, other) is not None + + asyncio.run(_go()) + + +def test_a_provider_account_s_role_is_not_edited_here(app): + from steward.models.users import User, UserRole + + async def _go(): + client, _ = await _client(app) + uid = str(uuid.uuid4()) + async with app.db_sessionmaker() as s: + async with s.begin(): + s.add(User(id=uid, username=f"sso-{uid[:8]}", email=f"{uid}@oidc", password_hash="x", + role=UserRole.viewer, auth_source="oidc", external_id=uid)) + resp = await client.post(f"/settings/users/{uid}/role", headers=CSRF, form={"role": "admin"}) + assert resp.status_code == 400 and "at each sign-in" in await resp.get_data(as_text=True) + resp = await client.post(f"/settings/users/{uid}/password", headers=CSRF, + form={"password": "long enough", "confirm": "long enough"}) + assert resp.status_code == 400 + assert (await _user(app, uid)).role.value == "viewer" + + asyncio.run(_go()) + + +def test_only_an_admin_reaches_users(app): + async def _go(): + client, _ = await _client(app, role="operator") + assert (await client.get("/settings/users/")).status_code == 403 + assert (await client.post("/settings/users/new", headers=CSRF, form={})).status_code == 403 + + asyncio.run(_go())