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//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//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//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//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'))