shopdb-flask/shopdb/core/api/users.py

"""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,
    guest_grantable_permissions)
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()
    # What a guest MAY be granted, so the grid can grey out the rest rather
    # than letting somebody tick a box the API will refuse.
    guestgrantable = sorted(guest_grantable_permissions())
    return success_response([{
        'roleid': r.roleid,
        'rolename': r.rolename,
        'description': r.description,
        'color': r.color,
        'usercount': r.users.count(),
        'permissions': [p.name for p in r.permissions],
        'isadmin': r.rolename == 'admin',
        'isguest': r.rolename == Role.GUEST,
        'grantable': guestgrantable if r.rolename == Role.GUEST else None,
    } 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'),
        color=data.get('color')
    )

    # 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,
        'color': role.color,
        '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']

    # Color is cosmetic - editable on any role, including admin
    if 'color' in data:
        if data['color'] != role.color:
            changes['color'] = {'old': role.color, 'new': data['color']}
        role.color = data['color']

    # Update permissions
    if 'permissions' in data and role.rolename != 'admin':
        # The guest role is anyone with no account, so what it may hold is
        # bounded: reads, plus the handful of writes a plugin has explicitly
        # declared safe to do anonymously. Refused at the door rather than
        # warned about - a checkbox that quietly makes deleting an asset
        # anonymous is not a warning anybody reads twice.
        if role.rolename == Role.GUEST:
            allowed = guest_grantable_permissions()
            refused = sorted(set(data['permissions']) - allowed)
            if refused:
                return error_response(
                    ErrorCodes.VALIDATION_ERROR,
                    'The guest role cannot hold {}. A guest may be granted any '
                    'view permission, and only writes a plugin has declared '
                    'safe to do without an account.'.format(', '.join(refused)))
        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,
        'color': role.color,
        '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')

    # Deleting the guest role does not close the door, it removes the only
    # place the door is described. Revoke its permissions instead.
    if role.rolename == Role.GUEST:
        return error_response(
            ErrorCodes.VALIDATION_ERROR,
            'Cannot delete the guest role. To close guest access, clear its '
            'permissions.')

    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, 'color': r.color} 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
    }