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