IT Assets: add role-based auth system and portal user sync
Auth: - Fix local login (was redirecting to portal; now authenticates AdminUser directly) - Portal SSO still takes priority in production via nginx headers Role system (admin | editor | readonly): - New app/utils/decorators.py with editor_required and admin_required decorators - All write routes protected with editor_required (create/edit/delete/import/mask) - Settings user management protected with admin_required - Sidebar hides write-only links for readonly users - Dashboard quick actions and list page buttons hidden for readonly Settings page: - Role colour badges (admin=red, editor=blue, readonly=grey) - Inline role changer per user (dropdown auto-submit) - Reset password modal per user - Delete user button with confirmation - Add user form includes role selector with legend Portal user sync: - New /internal/sync-user endpoint receives user pre-creation from portal - INTERNAL_SYNC_SECRET added to config - portal/config.py: added internal_url for itassets app so _sync_user_to_app works
This commit is contained in:
ske087
@@ -6,8 +6,9 @@ from app.routes.assignments import bp as assignments_bp
|
||||
from app.routes.paperwork import bp as paperwork_bp
|
||||
from app.routes.audit import bp as audit_bp
|
||||
from app.routes.settings import bp as settings_bp
|
||||
from app.routes.internal import bp as internal_bp
|
||||
|
||||
__all__ = [
|
||||
'auth_bp', 'dashboard_bp', 'users_bp', 'assets_bp',
|
||||
'assignments_bp', 'paperwork_bp', 'audit_bp', 'settings_bp',
|
||||
'assignments_bp', 'paperwork_bp', 'audit_bp', 'settings_bp', 'internal_bp',
|
||||
]
|
||||
|
||||
@@ -4,6 +4,7 @@ from flask import (Blueprint, render_template, redirect, url_for,
|
||||
flash, request, current_app, jsonify)
|
||||
from flask_login import login_required, current_user
|
||||
from app.extensions import db
|
||||
from app.utils.decorators import editor_required
|
||||
from app.models.asset import Asset, ASSET_TYPES, ASSET_STATUSES
|
||||
from app.models.audit_log import AuditLog
|
||||
from app.models.compliance_check import ComplianceCheck
|
||||
@@ -82,6 +83,7 @@ def index():
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/new', methods=['GET', 'POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def create():
|
||||
if request.method == 'POST':
|
||||
sn = request.form.get('serial_number', '').strip()
|
||||
@@ -204,6 +206,7 @@ def detail(asset_id):
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/<int:asset_id>/edit', methods=['GET', 'POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def edit(asset_id):
|
||||
asset = Asset.query.get_or_404(asset_id)
|
||||
|
||||
@@ -267,6 +270,7 @@ _COMPLIANCE_FIELDS = {
|
||||
|
||||
@bp.route('/<int:asset_id>/compliance', methods=['POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def update_compliance(asset_id):
|
||||
asset = Asset.query.get_or_404(asset_id)
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ 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.utils.decorators import editor_required
|
||||
from app.models.assignment import Assignment
|
||||
from app.models.asset import Asset
|
||||
from app.models.user import User
|
||||
@@ -46,6 +47,7 @@ def index():
|
||||
|
||||
@bp.route('/new', methods=['GET', 'POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def create():
|
||||
# Pre-fill from query params (used from asset / user detail pages)
|
||||
preselect_asset_id = request.args.get('asset_id', type=int)
|
||||
@@ -110,6 +112,7 @@ def create():
|
||||
|
||||
@bp.route('/<int:assignment_id>/return', methods=['POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def return_asset(assignment_id):
|
||||
assignment = Assignment.query.get_or_404(assignment_id)
|
||||
|
||||
|
||||
@@ -1,18 +1,36 @@
|
||||
from flask import Blueprint, redirect, url_for, current_app
|
||||
from flask_login import logout_user, login_required
|
||||
from datetime import datetime
|
||||
from flask import Blueprint, redirect, url_for, render_template, request, flash
|
||||
from flask_login import login_user, logout_user, login_required, current_user
|
||||
from app.extensions import db
|
||||
from app.models.admin_user import AdminUser
|
||||
|
||||
bp = Blueprint('auth', __name__)
|
||||
|
||||
|
||||
@bp.route('/login', methods=['GET', 'POST'])
|
||||
def login():
|
||||
# Authentication is handled by the EDP portal via nginx SSO.
|
||||
# Redirect any direct login attempts to the portal.
|
||||
return redirect(current_app.config.get('PORTAL_LOGIN_URL', 'http://localhost:8080/login'))
|
||||
# Portal SSO (nginx headers) handles production login automatically.
|
||||
# This route handles direct / standalone access.
|
||||
if current_user.is_authenticated:
|
||||
return redirect(url_for('dashboard.index'))
|
||||
|
||||
if request.method == 'POST':
|
||||
username = request.form.get('username', '').strip()
|
||||
password = request.form.get('password', '')
|
||||
user = AdminUser.query.filter_by(username=username, is_active=True).first()
|
||||
if user and user.check_password(password):
|
||||
login_user(user, remember=False)
|
||||
user.last_login = datetime.utcnow()
|
||||
db.session.commit()
|
||||
next_page = request.args.get('next') or url_for('dashboard.index')
|
||||
return redirect(next_page)
|
||||
flash('Invalid username or password.', 'danger')
|
||||
|
||||
return render_template('auth/login.html')
|
||||
|
||||
|
||||
@bp.route('/logout')
|
||||
@login_required
|
||||
def logout():
|
||||
logout_user()
|
||||
return redirect(current_app.config.get('PORTAL_LOGOUT_URL', 'http://localhost:8080/logout'))
|
||||
return redirect(url_for('auth.login'))
|
||||
|
||||
@@ -4,6 +4,7 @@ from flask import (Blueprint, render_template, redirect, url_for,
|
||||
flash, request, current_app, send_from_directory, jsonify)
|
||||
from flask_login import login_required, current_user
|
||||
from werkzeug.utils import secure_filename
|
||||
from app.utils.decorators import editor_required
|
||||
from app.extensions import db
|
||||
from app.models.document_template import DocumentTemplate
|
||||
from app.models.paperwork import DOC_TYPES
|
||||
@@ -54,6 +55,7 @@ def index():
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/upload', methods=['GET', 'POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def upload():
|
||||
if request.method == 'POST':
|
||||
name = request.form.get('name', '').strip()
|
||||
@@ -134,6 +136,7 @@ def download(tpl_id):
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/<int:tpl_id>/rescan', methods=['POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def rescan(tpl_id):
|
||||
tpl = DocumentTemplate.query.get_or_404(tpl_id)
|
||||
folder = _template_folder(current_app)
|
||||
@@ -153,6 +156,7 @@ def rescan(tpl_id):
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/<int:tpl_id>/edit', methods=['GET', 'POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def edit(tpl_id):
|
||||
tpl = DocumentTemplate.query.get_or_404(tpl_id)
|
||||
if request.method == 'POST':
|
||||
@@ -170,6 +174,7 @@ def edit(tpl_id):
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/<int:tpl_id>/delete', methods=['POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def delete(tpl_id):
|
||||
tpl = DocumentTemplate.query.get_or_404(tpl_id)
|
||||
# Check if any documents were generated from this template
|
||||
|
||||
@@ -0,0 +1,73 @@
|
||||
"""
|
||||
Internal routes — called by the EDP portal (not by browsers).
|
||||
|
||||
POST /internal/sync-user
|
||||
Called by the portal whenever it grants or updates a user's access to the
|
||||
itassets app. Ensures the local AdminUser record exists and has the correct
|
||||
role before the user ever logs in for the first time.
|
||||
|
||||
Body (JSON): { "username": "alice", "role": "advanced" }
|
||||
Header: X-Internal-Token: <INTERNAL_SYNC_SECRET>
|
||||
|
||||
Portal roles are mapped to local roles:
|
||||
admin → admin
|
||||
advanced → editor
|
||||
standard → readonly
|
||||
"""
|
||||
import secrets
|
||||
from flask import Blueprint, request, jsonify, current_app
|
||||
from app.extensions import db
|
||||
from app.models.admin_user import AdminUser
|
||||
from app.utils.portal_sso import _portal_role_to_local
|
||||
|
||||
bp = Blueprint('internal', __name__, url_prefix='/internal')
|
||||
|
||||
|
||||
def _check_token():
|
||||
expected = current_app.config.get('INTERNAL_SYNC_SECRET', '')
|
||||
provided = request.headers.get('X-Internal-Token', '')
|
||||
# Use hmac.compare_digest equivalent via secrets.compare_digest
|
||||
if not expected or not secrets.compare_digest(expected, provided):
|
||||
return False
|
||||
return True
|
||||
|
||||
|
||||
@bp.route('/sync-user', methods=['POST'])
|
||||
def sync_user():
|
||||
if not _check_token():
|
||||
return jsonify({'error': 'Unauthorized'}), 401
|
||||
|
||||
data = request.get_json(silent=True) or {}
|
||||
username = (data.get('username') or '').strip()
|
||||
portal_role = (data.get('role') or 'standard').strip()
|
||||
|
||||
if not username:
|
||||
return jsonify({'error': 'username is required'}), 400
|
||||
|
||||
local_role = _portal_role_to_local(portal_role)
|
||||
|
||||
try:
|
||||
user = AdminUser.query.filter_by(username=username).first()
|
||||
if user:
|
||||
if user.role != local_role:
|
||||
user.role = local_role
|
||||
db.session.commit()
|
||||
status = 'updated'
|
||||
else:
|
||||
user = AdminUser(
|
||||
username=username,
|
||||
email=f'{username}@portal.local',
|
||||
full_name=username,
|
||||
role=local_role,
|
||||
is_active=True,
|
||||
)
|
||||
user.set_password(secrets.token_hex(32)) # random unusable password
|
||||
db.session.add(user)
|
||||
db.session.commit()
|
||||
status = 'created'
|
||||
|
||||
return jsonify({'status': status, 'username': username, 'role': local_role}), 200
|
||||
|
||||
except Exception as exc:
|
||||
current_app.logger.error('sync-user failed: %s', exc)
|
||||
return jsonify({'error': str(exc)}), 500
|
||||
@@ -5,6 +5,7 @@ from flask import (Blueprint, render_template, redirect, url_for,
|
||||
flash, request, current_app, send_from_directory, abort, jsonify)
|
||||
from flask_login import login_required, current_user
|
||||
from app.extensions import db
|
||||
from app.utils.decorators import editor_required
|
||||
from app.models.paperwork import Paperwork, DOC_TYPES
|
||||
from app.models.document_template import DocumentTemplate
|
||||
from app.models.user import User
|
||||
@@ -54,6 +55,7 @@ def index():
|
||||
|
||||
@bp.route('/new', methods=['GET', 'POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def create():
|
||||
preselect_user_id = request.args.get('user_id', type=int)
|
||||
preselect_asset_id = request.args.get('asset_id', type=int)
|
||||
@@ -177,6 +179,7 @@ def download_docx(doc_id):
|
||||
|
||||
@bp.route('/<int:doc_id>/regenerate', methods=['POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def regenerate(doc_id):
|
||||
doc = Paperwork.query.get_or_404(doc_id)
|
||||
app = current_app._get_current_object()
|
||||
@@ -204,6 +207,7 @@ def regenerate(doc_id):
|
||||
|
||||
@bp.route('/<int:doc_id>/sign', methods=['POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def sign(doc_id):
|
||||
"""Record a signature on a document."""
|
||||
doc = Paperwork.query.get_or_404(doc_id)
|
||||
@@ -234,6 +238,7 @@ def sign(doc_id):
|
||||
|
||||
@bp.route('/<int:doc_id>/unsign', methods=['POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def unsign(doc_id):
|
||||
doc = Paperwork.query.get_or_404(doc_id)
|
||||
doc.signed_at = None
|
||||
|
||||
@@ -2,6 +2,7 @@ from flask import Blueprint, render_template, redirect, url_for, flash, request,
|
||||
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
|
||||
|
||||
bp = Blueprint('settings', __name__, url_prefix='/settings')
|
||||
|
||||
@@ -15,12 +16,16 @@ def index():
|
||||
|
||||
@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', 'admin')
|
||||
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')
|
||||
@@ -34,12 +39,13 @@ def create_admin():
|
||||
admin.set_password(password)
|
||||
db.session.add(admin)
|
||||
db.session.commit()
|
||||
flash(f'Admin user "{username}" created.', 'success')
|
||||
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:
|
||||
@@ -48,5 +54,53 @@ def toggle_admin(admin_id):
|
||||
admin.is_active = not admin.is_active
|
||||
db.session.commit()
|
||||
status = 'activated' if admin.is_active else 'deactivated'
|
||||
flash(f'Admin "{admin.username}" {status}.', 'success')
|
||||
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'))
|
||||
|
||||
@@ -4,6 +4,7 @@ from flask import (Blueprint, render_template, redirect, url_for,
|
||||
flash, request, current_app, jsonify)
|
||||
from flask_login import login_required, current_user
|
||||
from app.extensions import db
|
||||
from app.utils.decorators import editor_required
|
||||
from app.models.user import User
|
||||
from app.models.audit_log import AuditLog
|
||||
from app.services.csv_service import parse_users_csv
|
||||
@@ -63,6 +64,7 @@ def index():
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/new', methods=['GET', 'POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def create():
|
||||
if request.method == 'POST':
|
||||
windows_id = request.form.get('windows_id', '').strip()
|
||||
@@ -113,6 +115,7 @@ def detail(user_id):
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/<int:user_id>/edit', methods=['GET', 'POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def edit(user_id):
|
||||
user = User.query.get_or_404(user_id)
|
||||
|
||||
@@ -148,6 +151,7 @@ def edit(user_id):
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/<int:user_id>/mask', methods=['POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def mask(user_id):
|
||||
user = User.query.get_or_404(user_id)
|
||||
|
||||
@@ -191,6 +195,7 @@ def import_page():
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/import/csv', methods=['POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def import_csv():
|
||||
file = request.files.get('csv_file')
|
||||
if not file or not file.filename.endswith('.csv'):
|
||||
@@ -249,6 +254,7 @@ def import_csv():
|
||||
# ------------------------------------------------------------------
|
||||
@bp.route('/import/ldap', methods=['POST'])
|
||||
@login_required
|
||||
@editor_required
|
||||
def import_ldap():
|
||||
if not current_app.config.get('LDAP_SERVER'):
|
||||
flash('LDAP server is not configured. Update Settings first.', 'danger')
|
||||
|
||||
Reference in New Issue
Block a user