ske087
95ea6a3a35
Users are sourced exclusively from the portal (via SSO auto-create or the Sync from Portal button). The Add User form and create_admin route are removed to prevent out-of-sync local-only accounts. A footer hint now links admins to the portal settings page.
139 lines
4.9 KiB
Python
139 lines
4.9 KiB
Python
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/<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'))
|