Users: deleting a user clears their API tokens and detaches audit rows
Deleting any user who owned an API token or appeared in the audit log hit the users FK and 500ed - the import's 'importer' account being the guaranteed case (its PAT plus every audit row the import wrote). Tokens are revoked outright; audit history is kept but detached (userid NULL), so the trail survives the account.
This commit is contained in:
cproudlock
@@ -1,389 +1,399 @@
|
||||
"""User management API routes."""
|
||||
|
||||
from flask import Blueprint, request
|
||||
from flask_jwt_extended import jwt_required, current_user
|
||||
from werkzeug.security import generate_password_hash
|
||||
|
||||
from shopdb.extensions import db
|
||||
from shopdb.core.models import (
|
||||
User, Role, Permission, AuditLog, full_permission_catalog)
|
||||
from shopdb.utils.responses import success_response, error_response, ErrorCodes
|
||||
from shopdb.utils.authz import require_role
|
||||
|
||||
users_bp = Blueprint('users', __name__)
|
||||
|
||||
|
||||
@users_bp.route('', methods=['GET'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def list_users():
|
||||
"""List all users."""
|
||||
users = User.query.order_by(User.username).all()
|
||||
return success_response([user_to_dict(u) for u in users])
|
||||
|
||||
|
||||
@users_bp.route('/<int:userid>', methods=['GET'])
|
||||
@jwt_required()
|
||||
def get_user(userid: int):
|
||||
"""Get a single user."""
|
||||
# inline: decorators cannot express admin-or-self
|
||||
if not current_user.hasrole('admin') and current_user.userid != userid:
|
||||
return error_response(ErrorCodes.FORBIDDEN, 'Access denied', http_code=403)
|
||||
|
||||
user = db.session.get(User, userid)
|
||||
if not user:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'User not found', http_code=404)
|
||||
|
||||
return success_response(user_to_dict(user))
|
||||
|
||||
|
||||
@users_bp.route('', methods=['POST'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def create_user():
|
||||
"""Create a new user."""
|
||||
data = request.get_json()
|
||||
if not data:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Request body required')
|
||||
|
||||
# Validate required fields
|
||||
if not data.get('username'):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Username is required')
|
||||
if not data.get('email'):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Email is required')
|
||||
if not data.get('password'):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Password is required')
|
||||
|
||||
# Check uniqueness
|
||||
if User.query.filter_by(username=data['username']).first():
|
||||
return error_response(ErrorCodes.CONFLICT, 'Username already exists', http_code=409)
|
||||
if User.query.filter_by(email=data['email']).first():
|
||||
return error_response(ErrorCodes.CONFLICT, 'Email already exists', http_code=409)
|
||||
|
||||
# Admin-created accounts are forced to change the password on first login
|
||||
# unless the admin explicitly opts out.
|
||||
mustchange = data.get('mustchangepassword', True)
|
||||
|
||||
user = User(
|
||||
username=data['username'],
|
||||
email=data['email'],
|
||||
passwordhash=generate_password_hash(data['password']),
|
||||
firstname=data.get('firstname'),
|
||||
lastname=data.get('lastname'),
|
||||
isactive=data.get('isactive', True),
|
||||
mustchangepassword=bool(mustchange)
|
||||
)
|
||||
|
||||
# Assign roles
|
||||
role_ids = data.get('roles', [])
|
||||
if role_ids:
|
||||
roles = Role.query.filter(Role.roleid.in_(role_ids)).all()
|
||||
user.roles = roles
|
||||
|
||||
db.session.add(user)
|
||||
|
||||
# Audit log
|
||||
AuditLog.log('created', 'User', entityname=user.username)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
# Best-effort welcome email. The account exists regardless of mail outcome;
|
||||
# a failure is surfaced as a warning in the response, never a hard error.
|
||||
warning = None
|
||||
if data.get('sendwelcome', True) and user.email:
|
||||
sent = _send_welcome_email(user, data['password'])
|
||||
if not sent:
|
||||
warning = 'User created but the welcome email could not be sent.'
|
||||
|
||||
payload = user_to_dict(user)
|
||||
if warning:
|
||||
payload['warning'] = warning
|
||||
return success_response(payload, message='User created', http_code=201)
|
||||
|
||||
|
||||
def _send_welcome_email(user, temp_password):
|
||||
"""Send a new-user welcome email with sign-in details. Returns True on send.
|
||||
|
||||
Best-effort: any failure (including email being disabled) returns False so
|
||||
the caller can surface a soft warning without failing user creation.
|
||||
"""
|
||||
from shopdb.core.api.settings import get_cached_settings
|
||||
from shopdb.utils.mailer import render_email, send_email
|
||||
|
||||
settings = get_cached_settings() or {}
|
||||
facility = settings.get('facility_name') or 'ShopDB'
|
||||
base_url = (settings.get('site_base_url') or '').rstrip('/')
|
||||
login_link = f'{base_url}/login' if base_url else 'the ShopDB sign-in page'
|
||||
|
||||
body = (
|
||||
f'<p>An account has been created for you at <strong>{facility}</strong>.</p>'
|
||||
'<table style="border-collapse:collapse;font-size:14px;margin:12px 0;">'
|
||||
f'<tr><td style="padding:4px 12px 4px 0;color:#666;">Username</td>'
|
||||
f'<td><strong>{user.username}</strong></td></tr>'
|
||||
f'<tr><td style="padding:4px 12px 4px 0;color:#666;">Temporary password</td>'
|
||||
f'<td><code>{temp_password}</code></td></tr>'
|
||||
'</table>'
|
||||
f'<p>Sign in at {login_link}. You will be asked to set a new password '
|
||||
'the first time you log in.</p>'
|
||||
)
|
||||
html, text = render_email(f'Welcome to {facility}', body)
|
||||
return send_email(user.email, f'Your {facility} account', html, text=text)
|
||||
|
||||
|
||||
@users_bp.route('/<int:userid>', methods=['PUT'])
|
||||
@jwt_required()
|
||||
def update_user(userid: int):
|
||||
"""Update a user."""
|
||||
# inline: decorators cannot express admin-or-self
|
||||
if not current_user.hasrole('admin') and current_user.userid != userid:
|
||||
return error_response(ErrorCodes.FORBIDDEN, 'Access denied', http_code=403)
|
||||
|
||||
user = db.session.get(User, userid)
|
||||
if not user:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'User not found', http_code=404)
|
||||
|
||||
data = request.get_json()
|
||||
if not data:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Request body required')
|
||||
|
||||
changes = {}
|
||||
|
||||
# Update fields
|
||||
if 'email' in data and data['email'] != user.email:
|
||||
if User.query.filter(User.email == data['email'], User.userid != userid).first():
|
||||
return error_response(ErrorCodes.CONFLICT, 'Email already in use', http_code=409)
|
||||
changes['email'] = {'old': user.email, 'new': data['email']}
|
||||
user.email = data['email']
|
||||
|
||||
if 'firstname' in data:
|
||||
if data['firstname'] != user.firstname:
|
||||
changes['firstname'] = {'old': user.firstname, 'new': data['firstname']}
|
||||
user.firstname = data['firstname']
|
||||
|
||||
if 'lastname' in data:
|
||||
if data['lastname'] != user.lastname:
|
||||
changes['lastname'] = {'old': user.lastname, 'new': data['lastname']}
|
||||
user.lastname = data['lastname']
|
||||
|
||||
# Admin-only fields (inline: gates a subset of fields on a shared route)
|
||||
if current_user.hasrole('admin'):
|
||||
if 'isactive' in data:
|
||||
if data['isactive'] != user.isactive:
|
||||
changes['isactive'] = {'old': user.isactive, 'new': data['isactive']}
|
||||
user.isactive = data['isactive']
|
||||
|
||||
if 'roles' in data:
|
||||
old_roles = [r.rolename for r in user.roles]
|
||||
roles = Role.query.filter(Role.roleid.in_(data['roles'])).all()
|
||||
new_roles = [r.rolename for r in roles]
|
||||
if set(old_roles) != set(new_roles):
|
||||
changes['roles'] = {'old': old_roles, 'new': new_roles}
|
||||
user.roles = roles
|
||||
|
||||
# Unlock user
|
||||
if data.get('unlock'):
|
||||
user.lockeduntil = None
|
||||
user.failedlogins = 0
|
||||
changes['unlocked'] = {'old': True, 'new': False}
|
||||
|
||||
# Password change
|
||||
if 'password' in data and data['password']:
|
||||
user.passwordhash = generate_password_hash(data['password'])
|
||||
changes['password'] = {'old': '***', 'new': '***'}
|
||||
|
||||
if changes:
|
||||
AuditLog.log('updated', 'User', entityid=user.userid, entityname=user.username, changes=changes)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return success_response(user_to_dict(user), message='User updated')
|
||||
|
||||
|
||||
@users_bp.route('/<int:userid>', methods=['DELETE'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def delete_user(userid: int):
|
||||
"""Delete a user."""
|
||||
if current_user.userid == userid:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Cannot delete your own account')
|
||||
|
||||
user = db.session.get(User, userid)
|
||||
if not user:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'User not found', http_code=404)
|
||||
|
||||
username = user.username
|
||||
db.session.delete(user)
|
||||
|
||||
AuditLog.log('deleted', 'User', entityid=userid, entityname=username)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return success_response(None, message='User deleted')
|
||||
|
||||
|
||||
# Permissions endpoints
|
||||
@users_bp.route('/permissions', methods=['GET'])
|
||||
@jwt_required()
|
||||
def list_permissions():
|
||||
"""List assignable permissions grouped by category.
|
||||
|
||||
Driven by full_permission_catalog() (core plus ENABLED plugins) so a
|
||||
disabled plugin's permissions drop out of the role grid. Roles assign by
|
||||
name; the permissionid comes from the seeded Permission row when present.
|
||||
"""
|
||||
idbyname = {p.name: p.permissionid for p in Permission.query.all()}
|
||||
|
||||
catalog = full_permission_catalog()
|
||||
catalog.sort(key=lambda e: (e[2], e[0]))
|
||||
|
||||
grouped = {}
|
||||
flat = []
|
||||
for name, description, category in catalog:
|
||||
entry = {
|
||||
'permissionid': idbyname.get(name),
|
||||
'name': name,
|
||||
'description': description,
|
||||
}
|
||||
grouped.setdefault(category, []).append(entry)
|
||||
flat.append({**entry, 'category': category})
|
||||
|
||||
return success_response({'permissions': flat, 'grouped': grouped})
|
||||
|
||||
|
||||
# Roles endpoints
|
||||
@users_bp.route('/roles', methods=['GET'])
|
||||
@jwt_required()
|
||||
def list_roles():
|
||||
"""List all roles with their permissions."""
|
||||
roles = Role.query.order_by(Role.rolename).all()
|
||||
return success_response([{
|
||||
'roleid': r.roleid,
|
||||
'rolename': r.rolename,
|
||||
'description': r.description,
|
||||
'usercount': r.users.count(),
|
||||
'permissions': [p.name for p in r.permissions],
|
||||
'isadmin': r.rolename == 'admin'
|
||||
} for r in roles])
|
||||
|
||||
|
||||
@users_bp.route('/roles', methods=['POST'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def create_role():
|
||||
"""Create a new role."""
|
||||
data = request.get_json()
|
||||
if not data or not data.get('rolename'):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Role name is required')
|
||||
|
||||
if Role.query.filter_by(rolename=data['rolename']).first():
|
||||
return error_response(ErrorCodes.CONFLICT, 'Role already exists', http_code=409)
|
||||
|
||||
role = Role(
|
||||
rolename=data['rolename'],
|
||||
description=data.get('description')
|
||||
)
|
||||
|
||||
# Assign permissions
|
||||
if 'permissions' in data:
|
||||
perms = Permission.query.filter(Permission.name.in_(data['permissions'])).all()
|
||||
role.permissions = perms
|
||||
|
||||
db.session.add(role)
|
||||
|
||||
AuditLog.log('created', 'Role', entityname=role.rolename)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return success_response({
|
||||
'roleid': role.roleid,
|
||||
'rolename': role.rolename,
|
||||
'description': role.description,
|
||||
'permissions': [p.name for p in role.permissions]
|
||||
}, message='Role created', http_code=201)
|
||||
|
||||
|
||||
@users_bp.route('/roles/<int:roleid>', methods=['PUT'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def update_role(roleid: int):
|
||||
"""Update a role."""
|
||||
role = db.session.get(Role, roleid)
|
||||
if not role:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'Role not found', http_code=404)
|
||||
|
||||
# Cannot modify admin role permissions
|
||||
if role.rolename == 'admin' and 'permissions' in request.get_json():
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Cannot modify admin role permissions')
|
||||
|
||||
data = request.get_json()
|
||||
changes = {}
|
||||
|
||||
if 'description' in data:
|
||||
if data['description'] != role.description:
|
||||
changes['description'] = {'old': role.description, 'new': data['description']}
|
||||
role.description = data['description']
|
||||
|
||||
# Update permissions
|
||||
if 'permissions' in data and role.rolename != 'admin':
|
||||
old_perms = [p.name for p in role.permissions]
|
||||
perms = Permission.query.filter(Permission.name.in_(data['permissions'])).all()
|
||||
new_perms = [p.name for p in perms]
|
||||
if set(old_perms) != set(new_perms):
|
||||
changes['permissions'] = {'old': old_perms, 'new': new_perms}
|
||||
role.permissions = perms
|
||||
|
||||
if changes:
|
||||
AuditLog.log('updated', 'Role', entityid=role.roleid, entityname=role.rolename, changes=changes)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return success_response({
|
||||
'roleid': role.roleid,
|
||||
'rolename': role.rolename,
|
||||
'description': role.description,
|
||||
'permissions': [p.name for p in role.permissions]
|
||||
}, message='Role updated')
|
||||
|
||||
|
||||
@users_bp.route('/roles/<int:roleid>', methods=['DELETE'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def delete_role(roleid: int):
|
||||
"""Delete a role."""
|
||||
role = db.session.get(Role, roleid)
|
||||
if not role:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'Role not found', http_code=404)
|
||||
|
||||
if role.rolename == 'admin':
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Cannot delete the admin role')
|
||||
|
||||
if role.users.count() > 0:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, f'Role is assigned to {role.users.count()} user(s)')
|
||||
|
||||
rolename = role.rolename
|
||||
db.session.delete(role)
|
||||
|
||||
AuditLog.log('deleted', 'Role', entityid=roleid, entityname=rolename)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return success_response(None, message='Role deleted')
|
||||
|
||||
|
||||
def user_to_dict(user: User) -> dict:
|
||||
"""Convert user to dict for API response."""
|
||||
return {
|
||||
'userid': user.userid,
|
||||
'username': user.username,
|
||||
'email': user.email,
|
||||
'firstname': user.firstname,
|
||||
'lastname': user.lastname,
|
||||
'isactive': user.isactive,
|
||||
'islocked': user.islocked,
|
||||
'mustchangepassword': bool(user.mustchangepassword),
|
||||
'lastlogindate': user.lastlogindate.isoformat() + 'Z' if user.lastlogindate else None,
|
||||
'failedlogins': user.failedlogins,
|
||||
'roles': [{'roleid': r.roleid, 'rolename': r.rolename} for r in user.roles],
|
||||
'createddate': user.createddate.isoformat() + 'Z' if user.createddate else None,
|
||||
'modifieddate': user.modifieddate.isoformat() + 'Z' if user.modifieddate else None
|
||||
}
|
||||
"""User management API routes."""
|
||||
|
||||
from flask import Blueprint, request
|
||||
from flask_jwt_extended import jwt_required, current_user
|
||||
from werkzeug.security import generate_password_hash
|
||||
|
||||
from shopdb.extensions import db
|
||||
from shopdb.core.models import (
|
||||
User, Role, Permission, AuditLog, full_permission_catalog)
|
||||
from shopdb.utils.responses import success_response, error_response, ErrorCodes
|
||||
from shopdb.utils.authz import require_role
|
||||
|
||||
users_bp = Blueprint('users', __name__)
|
||||
|
||||
|
||||
@users_bp.route('', methods=['GET'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def list_users():
|
||||
"""List all users."""
|
||||
users = User.query.order_by(User.username).all()
|
||||
return success_response([user_to_dict(u) for u in users])
|
||||
|
||||
|
||||
@users_bp.route('/<int:userid>', methods=['GET'])
|
||||
@jwt_required()
|
||||
def get_user(userid: int):
|
||||
"""Get a single user."""
|
||||
# inline: decorators cannot express admin-or-self
|
||||
if not current_user.hasrole('admin') and current_user.userid != userid:
|
||||
return error_response(ErrorCodes.FORBIDDEN, 'Access denied', http_code=403)
|
||||
|
||||
user = db.session.get(User, userid)
|
||||
if not user:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'User not found', http_code=404)
|
||||
|
||||
return success_response(user_to_dict(user))
|
||||
|
||||
|
||||
@users_bp.route('', methods=['POST'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def create_user():
|
||||
"""Create a new user."""
|
||||
data = request.get_json()
|
||||
if not data:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Request body required')
|
||||
|
||||
# Validate required fields
|
||||
if not data.get('username'):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Username is required')
|
||||
if not data.get('email'):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Email is required')
|
||||
if not data.get('password'):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Password is required')
|
||||
|
||||
# Check uniqueness
|
||||
if User.query.filter_by(username=data['username']).first():
|
||||
return error_response(ErrorCodes.CONFLICT, 'Username already exists', http_code=409)
|
||||
if User.query.filter_by(email=data['email']).first():
|
||||
return error_response(ErrorCodes.CONFLICT, 'Email already exists', http_code=409)
|
||||
|
||||
# Admin-created accounts are forced to change the password on first login
|
||||
# unless the admin explicitly opts out.
|
||||
mustchange = data.get('mustchangepassword', True)
|
||||
|
||||
user = User(
|
||||
username=data['username'],
|
||||
email=data['email'],
|
||||
passwordhash=generate_password_hash(data['password']),
|
||||
firstname=data.get('firstname'),
|
||||
lastname=data.get('lastname'),
|
||||
isactive=data.get('isactive', True),
|
||||
mustchangepassword=bool(mustchange)
|
||||
)
|
||||
|
||||
# Assign roles
|
||||
role_ids = data.get('roles', [])
|
||||
if role_ids:
|
||||
roles = Role.query.filter(Role.roleid.in_(role_ids)).all()
|
||||
user.roles = roles
|
||||
|
||||
db.session.add(user)
|
||||
|
||||
# Audit log
|
||||
AuditLog.log('created', 'User', entityname=user.username)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
# Best-effort welcome email. The account exists regardless of mail outcome;
|
||||
# a failure is surfaced as a warning in the response, never a hard error.
|
||||
warning = None
|
||||
if data.get('sendwelcome', True) and user.email:
|
||||
sent = _send_welcome_email(user, data['password'])
|
||||
if not sent:
|
||||
warning = 'User created but the welcome email could not be sent.'
|
||||
|
||||
payload = user_to_dict(user)
|
||||
if warning:
|
||||
payload['warning'] = warning
|
||||
return success_response(payload, message='User created', http_code=201)
|
||||
|
||||
|
||||
def _send_welcome_email(user, temp_password):
|
||||
"""Send a new-user welcome email with sign-in details. Returns True on send.
|
||||
|
||||
Best-effort: any failure (including email being disabled) returns False so
|
||||
the caller can surface a soft warning without failing user creation.
|
||||
"""
|
||||
from shopdb.core.api.settings import get_cached_settings
|
||||
from shopdb.utils.mailer import render_email, send_email
|
||||
|
||||
settings = get_cached_settings() or {}
|
||||
facility = settings.get('facility_name') or 'ShopDB'
|
||||
base_url = (settings.get('site_base_url') or '').rstrip('/')
|
||||
login_link = f'{base_url}/login' if base_url else 'the ShopDB sign-in page'
|
||||
|
||||
body = (
|
||||
f'<p>An account has been created for you at <strong>{facility}</strong>.</p>'
|
||||
'<table style="border-collapse:collapse;font-size:14px;margin:12px 0;">'
|
||||
f'<tr><td style="padding:4px 12px 4px 0;color:#666;">Username</td>'
|
||||
f'<td><strong>{user.username}</strong></td></tr>'
|
||||
f'<tr><td style="padding:4px 12px 4px 0;color:#666;">Temporary password</td>'
|
||||
f'<td><code>{temp_password}</code></td></tr>'
|
||||
'</table>'
|
||||
f'<p>Sign in at {login_link}. You will be asked to set a new password '
|
||||
'the first time you log in.</p>'
|
||||
)
|
||||
html, text = render_email(f'Welcome to {facility}', body)
|
||||
return send_email(user.email, f'Your {facility} account', html, text=text)
|
||||
|
||||
|
||||
@users_bp.route('/<int:userid>', methods=['PUT'])
|
||||
@jwt_required()
|
||||
def update_user(userid: int):
|
||||
"""Update a user."""
|
||||
# inline: decorators cannot express admin-or-self
|
||||
if not current_user.hasrole('admin') and current_user.userid != userid:
|
||||
return error_response(ErrorCodes.FORBIDDEN, 'Access denied', http_code=403)
|
||||
|
||||
user = db.session.get(User, userid)
|
||||
if not user:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'User not found', http_code=404)
|
||||
|
||||
data = request.get_json()
|
||||
if not data:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Request body required')
|
||||
|
||||
changes = {}
|
||||
|
||||
# Update fields
|
||||
if 'email' in data and data['email'] != user.email:
|
||||
if User.query.filter(User.email == data['email'], User.userid != userid).first():
|
||||
return error_response(ErrorCodes.CONFLICT, 'Email already in use', http_code=409)
|
||||
changes['email'] = {'old': user.email, 'new': data['email']}
|
||||
user.email = data['email']
|
||||
|
||||
if 'firstname' in data:
|
||||
if data['firstname'] != user.firstname:
|
||||
changes['firstname'] = {'old': user.firstname, 'new': data['firstname']}
|
||||
user.firstname = data['firstname']
|
||||
|
||||
if 'lastname' in data:
|
||||
if data['lastname'] != user.lastname:
|
||||
changes['lastname'] = {'old': user.lastname, 'new': data['lastname']}
|
||||
user.lastname = data['lastname']
|
||||
|
||||
# Admin-only fields (inline: gates a subset of fields on a shared route)
|
||||
if current_user.hasrole('admin'):
|
||||
if 'isactive' in data:
|
||||
if data['isactive'] != user.isactive:
|
||||
changes['isactive'] = {'old': user.isactive, 'new': data['isactive']}
|
||||
user.isactive = data['isactive']
|
||||
|
||||
if 'roles' in data:
|
||||
old_roles = [r.rolename for r in user.roles]
|
||||
roles = Role.query.filter(Role.roleid.in_(data['roles'])).all()
|
||||
new_roles = [r.rolename for r in roles]
|
||||
if set(old_roles) != set(new_roles):
|
||||
changes['roles'] = {'old': old_roles, 'new': new_roles}
|
||||
user.roles = roles
|
||||
|
||||
# Unlock user
|
||||
if data.get('unlock'):
|
||||
user.lockeduntil = None
|
||||
user.failedlogins = 0
|
||||
changes['unlocked'] = {'old': True, 'new': False}
|
||||
|
||||
# Password change
|
||||
if 'password' in data and data['password']:
|
||||
user.passwordhash = generate_password_hash(data['password'])
|
||||
changes['password'] = {'old': '***', 'new': '***'}
|
||||
|
||||
if changes:
|
||||
AuditLog.log('updated', 'User', entityid=user.userid, entityname=user.username, changes=changes)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return success_response(user_to_dict(user), message='User updated')
|
||||
|
||||
|
||||
@users_bp.route('/<int:userid>', methods=['DELETE'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def delete_user(userid: int):
|
||||
"""Delete a user."""
|
||||
if current_user.userid == userid:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Cannot delete your own account')
|
||||
|
||||
user = db.session.get(User, userid)
|
||||
if not user:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'User not found', http_code=404)
|
||||
|
||||
username = user.username
|
||||
|
||||
# Rows that reference the user would otherwise block the delete:
|
||||
# revoke their API tokens outright, and DETACH their audit history
|
||||
# (userid -> NULL) - the log rows themselves are kept, entityname and
|
||||
# details still tell the story.
|
||||
from shopdb.core.models import ApiToken
|
||||
ApiToken.query.filter_by(userid=userid).delete(synchronize_session=False)
|
||||
AuditLog.query.filter_by(userid=userid).update(
|
||||
{'userid': None}, synchronize_session=False)
|
||||
|
||||
db.session.delete(user)
|
||||
|
||||
AuditLog.log('deleted', 'User', entityid=userid, entityname=username)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return success_response(None, message='User deleted')
|
||||
|
||||
|
||||
# Permissions endpoints
|
||||
@users_bp.route('/permissions', methods=['GET'])
|
||||
@jwt_required()
|
||||
def list_permissions():
|
||||
"""List assignable permissions grouped by category.
|
||||
|
||||
Driven by full_permission_catalog() (core plus ENABLED plugins) so a
|
||||
disabled plugin's permissions drop out of the role grid. Roles assign by
|
||||
name; the permissionid comes from the seeded Permission row when present.
|
||||
"""
|
||||
idbyname = {p.name: p.permissionid for p in Permission.query.all()}
|
||||
|
||||
catalog = full_permission_catalog()
|
||||
catalog.sort(key=lambda e: (e[2], e[0]))
|
||||
|
||||
grouped = {}
|
||||
flat = []
|
||||
for name, description, category in catalog:
|
||||
entry = {
|
||||
'permissionid': idbyname.get(name),
|
||||
'name': name,
|
||||
'description': description,
|
||||
}
|
||||
grouped.setdefault(category, []).append(entry)
|
||||
flat.append({**entry, 'category': category})
|
||||
|
||||
return success_response({'permissions': flat, 'grouped': grouped})
|
||||
|
||||
|
||||
# Roles endpoints
|
||||
@users_bp.route('/roles', methods=['GET'])
|
||||
@jwt_required()
|
||||
def list_roles():
|
||||
"""List all roles with their permissions."""
|
||||
roles = Role.query.order_by(Role.rolename).all()
|
||||
return success_response([{
|
||||
'roleid': r.roleid,
|
||||
'rolename': r.rolename,
|
||||
'description': r.description,
|
||||
'usercount': r.users.count(),
|
||||
'permissions': [p.name for p in r.permissions],
|
||||
'isadmin': r.rolename == 'admin'
|
||||
} for r in roles])
|
||||
|
||||
|
||||
@users_bp.route('/roles', methods=['POST'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def create_role():
|
||||
"""Create a new role."""
|
||||
data = request.get_json()
|
||||
if not data or not data.get('rolename'):
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Role name is required')
|
||||
|
||||
if Role.query.filter_by(rolename=data['rolename']).first():
|
||||
return error_response(ErrorCodes.CONFLICT, 'Role already exists', http_code=409)
|
||||
|
||||
role = Role(
|
||||
rolename=data['rolename'],
|
||||
description=data.get('description')
|
||||
)
|
||||
|
||||
# Assign permissions
|
||||
if 'permissions' in data:
|
||||
perms = Permission.query.filter(Permission.name.in_(data['permissions'])).all()
|
||||
role.permissions = perms
|
||||
|
||||
db.session.add(role)
|
||||
|
||||
AuditLog.log('created', 'Role', entityname=role.rolename)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return success_response({
|
||||
'roleid': role.roleid,
|
||||
'rolename': role.rolename,
|
||||
'description': role.description,
|
||||
'permissions': [p.name for p in role.permissions]
|
||||
}, message='Role created', http_code=201)
|
||||
|
||||
|
||||
@users_bp.route('/roles/<int:roleid>', methods=['PUT'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def update_role(roleid: int):
|
||||
"""Update a role."""
|
||||
role = db.session.get(Role, roleid)
|
||||
if not role:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'Role not found', http_code=404)
|
||||
|
||||
# Cannot modify admin role permissions
|
||||
if role.rolename == 'admin' and 'permissions' in request.get_json():
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Cannot modify admin role permissions')
|
||||
|
||||
data = request.get_json()
|
||||
changes = {}
|
||||
|
||||
if 'description' in data:
|
||||
if data['description'] != role.description:
|
||||
changes['description'] = {'old': role.description, 'new': data['description']}
|
||||
role.description = data['description']
|
||||
|
||||
# Update permissions
|
||||
if 'permissions' in data and role.rolename != 'admin':
|
||||
old_perms = [p.name for p in role.permissions]
|
||||
perms = Permission.query.filter(Permission.name.in_(data['permissions'])).all()
|
||||
new_perms = [p.name for p in perms]
|
||||
if set(old_perms) != set(new_perms):
|
||||
changes['permissions'] = {'old': old_perms, 'new': new_perms}
|
||||
role.permissions = perms
|
||||
|
||||
if changes:
|
||||
AuditLog.log('updated', 'Role', entityid=role.roleid, entityname=role.rolename, changes=changes)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return success_response({
|
||||
'roleid': role.roleid,
|
||||
'rolename': role.rolename,
|
||||
'description': role.description,
|
||||
'permissions': [p.name for p in role.permissions]
|
||||
}, message='Role updated')
|
||||
|
||||
|
||||
@users_bp.route('/roles/<int:roleid>', methods=['DELETE'])
|
||||
@jwt_required()
|
||||
@require_role('admin')
|
||||
def delete_role(roleid: int):
|
||||
"""Delete a role."""
|
||||
role = db.session.get(Role, roleid)
|
||||
if not role:
|
||||
return error_response(ErrorCodes.NOT_FOUND, 'Role not found', http_code=404)
|
||||
|
||||
if role.rolename == 'admin':
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, 'Cannot delete the admin role')
|
||||
|
||||
if role.users.count() > 0:
|
||||
return error_response(ErrorCodes.VALIDATION_ERROR, f'Role is assigned to {role.users.count()} user(s)')
|
||||
|
||||
rolename = role.rolename
|
||||
db.session.delete(role)
|
||||
|
||||
AuditLog.log('deleted', 'Role', entityid=roleid, entityname=rolename)
|
||||
|
||||
db.session.commit()
|
||||
|
||||
return success_response(None, message='Role deleted')
|
||||
|
||||
|
||||
def user_to_dict(user: User) -> dict:
|
||||
"""Convert user to dict for API response."""
|
||||
return {
|
||||
'userid': user.userid,
|
||||
'username': user.username,
|
||||
'email': user.email,
|
||||
'firstname': user.firstname,
|
||||
'lastname': user.lastname,
|
||||
'isactive': user.isactive,
|
||||
'islocked': user.islocked,
|
||||
'mustchangepassword': bool(user.mustchangepassword),
|
||||
'lastlogindate': user.lastlogindate.isoformat() + 'Z' if user.lastlogindate else None,
|
||||
'failedlogins': user.failedlogins,
|
||||
'roles': [{'roleid': r.roleid, 'rolename': r.rolename} for r in user.roles],
|
||||
'createddate': user.createddate.isoformat() + 'Z' if user.createddate else None,
|
||||
'modifieddate': user.modifieddate.isoformat() + 'Z' if user.modifieddate else None
|
||||
}
|
||||
|
||||
29
tests/test_core/test_users_api.py
Normal file
29
tests/test_core/test_users_api.py
Normal file
@@ -0,0 +1,29 @@
|
||||
|
||||
|
||||
def test_delete_user_with_tokens_and_audit_history(client, auth_headers, app, db):
|
||||
"""Deleting a user revokes their API tokens and detaches (not deletes)
|
||||
their audit rows - the importer-user case."""
|
||||
from werkzeug.security import generate_password_hash
|
||||
from shopdb.core.models import User, ApiToken, AuditLog
|
||||
|
||||
with app.app_context():
|
||||
user = User(username='importer2', email='importer2@test.local',
|
||||
passwordhash=generate_password_hash('x'), isactive=True)
|
||||
db.session.add(user)
|
||||
db.session.flush()
|
||||
db.session.add(ApiToken(userid=user.userid, name='import token',
|
||||
tokenprefix='deadbeef', tokenhash='x' * 64))
|
||||
db.session.add(AuditLog(userid=user.userid, action='created',
|
||||
entitytype='Asset', entityid=1,
|
||||
entityname='imported thing'))
|
||||
db.session.commit()
|
||||
userid = user.userid
|
||||
|
||||
response = client.delete(f'/api/users/{userid}', headers=auth_headers)
|
||||
assert response.status_code == 200, response.get_json()
|
||||
|
||||
with app.app_context():
|
||||
assert db.session.get(User, userid) is None
|
||||
assert ApiToken.query.filter_by(userid=userid).count() == 0
|
||||
detached = AuditLog.query.filter_by(entityname='imported thing').one()
|
||||
assert detached.userid is None
|
||||
Reference in New Issue
Block a user