mteehan
a82578f342
- update_permissions now calls blacklist_active_refresh_token and rotate_user_secret to immediately invalidate stale tokens - create_user uses returned id from tx.run_one instead of redundant SELECT - websocket reconnect explicitly closes old connection after token refresh to prevent onclose handler race condition
280 lines
7.6 KiB
Python
280 lines
7.6 KiB
Python
"""User CRUD and permission management for Vacuum Wall authentication.
|
|
|
|
All operations use DB query IDs through the Database abstract layer.
|
|
Password hashing uses Argon2id via lib.password.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
import re
|
|
import secrets
|
|
from typing import Any
|
|
|
|
from lib.auth import (
|
|
blacklist_active_refresh_token,
|
|
blacklist_expired,
|
|
rotate_user_secret,
|
|
)
|
|
from lib.db import (
|
|
Q_DELETE_PERMISSIONS,
|
|
Q_DELETE_USER,
|
|
Q_INSERT_USER,
|
|
Q_SELECT_ALL_USERS,
|
|
Q_SELECT_PERMISSIONS,
|
|
Q_SELECT_USER_BY_NAME,
|
|
Q_UPDATE_PASSWORD,
|
|
Q_UPSERT_PERMISSION,
|
|
get_db,
|
|
)
|
|
from lib.password import hash_password, needs_rehash, verify_password
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
# All subsystem names for default permission assignment
|
|
ALL_SUBSYSTEMS = [
|
|
"firewall",
|
|
"network",
|
|
"dhcp",
|
|
"proxy",
|
|
"certs",
|
|
"wireguard",
|
|
"logs",
|
|
"status",
|
|
"auth",
|
|
]
|
|
|
|
_USERNAME_RE = re.compile(r"^[a-zA-Z0-9_-]{3,32}$")
|
|
|
|
|
|
def _get_permissions(username: str) -> dict[str, str]:
|
|
"""Load permissions for *username* from the database.
|
|
|
|
Args:
|
|
username: The username to load permissions for.
|
|
|
|
Returns:
|
|
Dict mapping subsystem names to permission levels.
|
|
"""
|
|
db = get_db()
|
|
rows = db.query(Q_SELECT_PERMISSIONS, (username,))
|
|
return {row["subsystem"]: row["level"] for row in rows}
|
|
|
|
|
|
def get_user(username: str) -> dict[str, Any] | None:
|
|
"""Get a user by username (without password hash or JWT secret).
|
|
|
|
Args:
|
|
username: The username to look up.
|
|
|
|
Returns:
|
|
User dict with id, username, permissions, or None if not found.
|
|
"""
|
|
user = find_user(username)
|
|
if user is None:
|
|
return None
|
|
return {
|
|
"id": user["id"],
|
|
"username": user["username"],
|
|
"permissions": _get_permissions(username),
|
|
}
|
|
|
|
|
|
def find_user(username: str) -> dict[str, Any] | None:
|
|
"""Find a user by username, including password hash and JWT secret.
|
|
|
|
Used for password verification and token operations. Not returned through APIs.
|
|
|
|
Args:
|
|
username: The username to look up.
|
|
|
|
Returns:
|
|
User dict from the database row, or None.
|
|
"""
|
|
db = get_db()
|
|
rows = db.query(Q_SELECT_USER_BY_NAME, (username,))
|
|
if not rows:
|
|
return None
|
|
return rows[0]
|
|
|
|
|
|
def verify_user_password(username: str, password: str) -> dict[str, Any] | None:
|
|
"""Verify a user's password using Argon2id.
|
|
|
|
Args:
|
|
username: The username to verify.
|
|
password: Plain-text password.
|
|
|
|
Returns:
|
|
User dict (without password hash) if the password is correct, None otherwise.
|
|
"""
|
|
user = find_user(username)
|
|
if user is None:
|
|
# Run a dummy Argon2id verification to prevent timing-based user enumeration.
|
|
# The timing for both paths is now equivalent.
|
|
verify_password(password, hash_password(secrets.token_hex(32)))
|
|
return None
|
|
if not verify_password(password, user["password_hash"]):
|
|
return None
|
|
if needs_rehash(user["password_hash"]):
|
|
new_hash = hash_password(password)
|
|
db = get_db()
|
|
db.run(Q_UPDATE_PASSWORD, (new_hash, username))
|
|
logger.info("Password hash rehashed for %r (param upgrade)", username)
|
|
return {
|
|
"id": user["id"],
|
|
"username": user["username"],
|
|
"permissions": _get_permissions(username),
|
|
}
|
|
|
|
|
|
def create_user(
|
|
username: str,
|
|
password: str,
|
|
permissions: dict[str, str] | None = None,
|
|
) -> dict[str, Any]:
|
|
"""Create a new user with the given password and permissions.
|
|
|
|
Args:
|
|
username: The new username (3-32 alphanumeric chars, dash, underscore).
|
|
password: Plain-text password that will be hashed with Argon2id.
|
|
permissions: Dict mapping subsystem names to "read" or "rw".
|
|
|
|
Returns:
|
|
The created user dict with id, username, permissions.
|
|
|
|
Raises:
|
|
ValueError: If username is invalid or already exists.
|
|
"""
|
|
if not _USERNAME_RE.match(username):
|
|
raise ValueError(
|
|
"Username must be 3-32 characters: letters, digits, dash, underscore"
|
|
)
|
|
|
|
existing = find_user(username)
|
|
if existing is not None:
|
|
raise ValueError(f"User {username!r} already exists")
|
|
|
|
password_hash = hash_password(password)
|
|
jwt_secret = secrets.token_urlsafe(32)
|
|
|
|
db = get_db()
|
|
with db.in_transaction() as tx:
|
|
user_id = tx.run_one(Q_INSERT_USER, (username, password_hash, jwt_secret))
|
|
if permissions:
|
|
for subsystem, level in permissions.items():
|
|
tx.run(Q_UPSERT_PERMISSION, (username, subsystem, level))
|
|
|
|
return {
|
|
"id": user_id,
|
|
"username": username,
|
|
"permissions": _get_permissions(username),
|
|
}
|
|
|
|
|
|
def update_password(username: str, old_password: str, new_password: str) -> bool:
|
|
"""Update a user's password and invalidate all active tokens.
|
|
|
|
Rotates the user's JWT secret, immediately invalidating all existing
|
|
access and refresh tokens.
|
|
|
|
Args:
|
|
username: The username.
|
|
old_password: Current password.
|
|
new_password: New plain-text password.
|
|
|
|
Returns:
|
|
True if password was updated.
|
|
|
|
Raises:
|
|
ValueError: If old password is incorrect.
|
|
"""
|
|
if not verify_user_password(username, old_password):
|
|
raise ValueError("Current password is incorrect")
|
|
|
|
blacklist_active_refresh_token(username)
|
|
new_hash = hash_password(new_password)
|
|
rotate_user_secret(username)
|
|
db = get_db()
|
|
db.run(Q_UPDATE_PASSWORD, (new_hash, username))
|
|
_cleanup_blacklist()
|
|
return True
|
|
|
|
|
|
def update_permissions(username: str, permissions: dict[str, str]) -> None:
|
|
"""Update a user's permissions and invalidate all existing tokens.
|
|
|
|
Replaces all existing permissions with the provided mapping. Rotates the
|
|
JWT secret so that permission changes take effect immediately — existing
|
|
tokens with stale permissions are no longer valid.
|
|
|
|
Args:
|
|
username: The username.
|
|
permissions: Dict mapping subsystem names to permission levels.
|
|
"""
|
|
user = find_user(username)
|
|
if user is None:
|
|
raise ValueError(f"User {username!r} not found")
|
|
|
|
db = get_db()
|
|
with db.in_transaction() as tx:
|
|
tx.run(Q_DELETE_PERMISSIONS, (username,))
|
|
for subsystem, level in permissions.items():
|
|
tx.run(Q_UPSERT_PERMISSION, (username, subsystem, level))
|
|
blacklist_active_refresh_token(username)
|
|
rotate_user_secret(username)
|
|
|
|
|
|
def list_users() -> list[dict[str, Any]]:
|
|
"""List all users with their permissions.
|
|
|
|
Returns:
|
|
List of user summary dicts.
|
|
"""
|
|
db = get_db()
|
|
rows = db.query(Q_SELECT_ALL_USERS, ())
|
|
|
|
result = []
|
|
for row in rows:
|
|
username = row["username"]
|
|
permissions = _get_permissions(username)
|
|
result.append(
|
|
{
|
|
"id": row["id"],
|
|
"username": username,
|
|
"permissions": permissions,
|
|
"created_at": row["created_at"],
|
|
}
|
|
)
|
|
return result
|
|
|
|
|
|
def delete_user(username: str) -> bool:
|
|
"""Delete a user and all their associated data.
|
|
|
|
Permissions and WebAuthn credentials are CASCADE-deleted by the schema.
|
|
|
|
Args:
|
|
username: The username to delete.
|
|
|
|
Returns:
|
|
True if the user was deleted.
|
|
|
|
Raises:
|
|
ValueError: If the user does not exist.
|
|
"""
|
|
user = find_user(username)
|
|
if user is None:
|
|
raise ValueError(f"User {username!r} not found")
|
|
|
|
blacklist_active_refresh_token(username)
|
|
db = get_db()
|
|
db.run(Q_DELETE_USER, (username,))
|
|
_cleanup_blacklist()
|
|
return True
|
|
|
|
|
|
def _cleanup_blacklist() -> None:
|
|
"""Clean up expired blacklist entries."""
|
|
blacklist_expired()
|