enterprise_digital-platform/IT_asset_management/app/routes/settings.py

from flask import Blueprint, render_template, redirect, url_for, flash, request, current_app
from flask_login import login_required, current_user
from app.extensions import db
from app.models.admin_user import AdminUser
from app.utils.decorators import admin_required
from app.utils.portal_sso import _portal_role_to_local

bp = Blueprint('settings', __name__, url_prefix='/settings')


@bp.route('/')
@login_required
def index():
    admins = AdminUser.query.order_by(AdminUser.username).all()
    return render_template('settings/index.html', admins=admins, config=current_app.config)


@bp.route('/admin/new', methods=['POST'])
@login_required
@admin_required
def create_admin():
    username = request.form.get('username', '').strip()
    email = request.form.get('email', '').strip()
    full_name = request.form.get('full_name', '').strip()
    password = request.form.get('password', '')
    role = request.form.get('role', 'readonly')

    if role not in ('admin', 'editor', 'readonly'):
        role = 'readonly'

    if not username or not email or not password:
        flash('Username, email and password are required.', 'danger')
        return redirect(url_for('settings.index'))

    if AdminUser.query.filter_by(username=username).first():
        flash(f'Username "{username}" is already taken.', 'danger')
        return redirect(url_for('settings.index'))

    admin = AdminUser(username=username, email=email, full_name=full_name, role=role)
    admin.set_password(password)
    db.session.add(admin)
    db.session.commit()
    flash(f'User "{username}" created with role "{role}".', 'success')
    return redirect(url_for('settings.index'))


@bp.route('/admin/<int:admin_id>/toggle', methods=['POST'])
@login_required
@admin_required
def toggle_admin(admin_id):
    admin = AdminUser.query.get_or_404(admin_id)
    if admin.id == current_user.id:
        flash('You cannot deactivate your own account.', 'danger')
        return redirect(url_for('settings.index'))
    admin.is_active = not admin.is_active
    db.session.commit()
    status = 'activated' if admin.is_active else 'deactivated'
    flash(f'User "{admin.username}" {status}.', 'success')
    return redirect(url_for('settings.index'))


@bp.route('/admin/<int:admin_id>/change-role', methods=['POST'])
@login_required
@admin_required
def change_role(admin_id):
    admin = AdminUser.query.get_or_404(admin_id)
    if admin.id == current_user.id:
        flash('You cannot change your own role.', 'danger')
        return redirect(url_for('settings.index'))
    role = request.form.get('role', 'readonly')
    if role not in ('admin', 'editor', 'readonly'):
        flash('Invalid role.', 'danger')
        return redirect(url_for('settings.index'))
    admin.role = role
    db.session.commit()
    flash(f'Role for "{admin.username}" updated to "{role}".', 'success')
    return redirect(url_for('settings.index'))


@bp.route('/admin/<int:admin_id>/reset-password', methods=['POST'])
@login_required
@admin_required
def reset_password(admin_id):
    admin = AdminUser.query.get_or_404(admin_id)
    new_password = request.form.get('new_password', '')
    if len(new_password) < 8:
        flash('Password must be at least 8 characters.', 'danger')
        return redirect(url_for('settings.index'))
    admin.set_password(new_password)
    db.session.commit()
    flash(f'Password for "{admin.username}" has been reset.', 'success')
    return redirect(url_for('settings.index'))


@bp.route('/admin/<int:admin_id>/delete', methods=['POST'])
@login_required
@admin_required
def delete_admin(admin_id):
    admin = AdminUser.query.get_or_404(admin_id)
    if admin.id == current_user.id:
        flash('You cannot delete your own account.', 'danger')
        return redirect(url_for('settings.index'))
    username = admin.username
    db.session.delete(admin)
    db.session.commit()
    flash(f'User "{username}" deleted.', 'success')
    return redirect(url_for('settings.index'))


@bp.route('/sync-from-portal', methods=['POST'])
@login_required
@admin_required
def sync_from_portal():
    """Pull all portal users with itassets access and upsert them locally."""
    import urllib.request
    import json as _json

    portal_url = current_app.config.get('PORTAL_INTERNAL_URL', 'http://localhost:5001')
    secret = current_app.config.get('INTERNAL_SYNC_SECRET', '')
    url = portal_url.rstrip('/') + '/api/internal/itassets-users'

    try:
        req = urllib.request.Request(
            url,
            headers={'X-Internal-Token': secret},
            method='GET',
        )
        with urllib.request.urlopen(req, timeout=5) as resp:
            users = _json.loads(resp.read())
    except Exception as exc:
        flash(f'Could not reach portal: {exc}', 'danger')
        return redirect(url_for('settings.index'))

    created = updated = 0
    for u in users:
        username = u.get('username', '').strip()
        email = u.get('email', '') or f'{username}@portal.local'
        local_role = _portal_role_to_local(u.get('app_role', 'standard'))
        if not username:
            continue
        existing = AdminUser.query.filter_by(username=username).first()
        if existing:
            changed = False
            if existing.role != local_role:
                existing.role = local_role
                changed = True
            if existing.email != email:
                existing.email = email
                changed = True
            if changed:
                updated += 1
        else:
            import secrets as _sec
            new_user = AdminUser(
                username=username,
                email=email,
                full_name=username,
                role=local_role,
                is_active=True,
            )
            new_user.set_password(_sec.token_hex(32))
            db.session.add(new_user)
            created += 1

    db.session.commit()
    flash(f'Portal sync complete — {created} created, {updated} updated.', 'success')
    return redirect(url_for('settings.index'))