321 lines
12 KiB
Python
321 lines
12 KiB
Python
from flask import Blueprint, render_template, redirect, url_for, flash, request, abort
|
|
from flask_login import login_user, logout_user, login_required, current_user
|
|
from app import db, limiter
|
|
from app.models.user import User
|
|
from app.utils.forms import LoginForm, UserForm, ProfileForm
|
|
from app.utils.decorators import admin_required, supervisor_required, safe_redirect_url
|
|
import logging
|
|
from app.utils.audit import log_action, ACTION_CREATE, ACTION_UPDATE, ACTION_DELETE, ACTION_LOGIN, ACTION_LOGOUT
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
bp = Blueprint('auth', __name__, url_prefix='/auth')
|
|
|
|
|
|
@bp.route('/login', methods=['GET', 'POST'])
|
|
@limiter.limit('20 per minute; 5 per second')
|
|
def login():
|
|
if current_user.is_authenticated:
|
|
return redirect(url_for('dashboard.index'))
|
|
|
|
form = LoginForm()
|
|
if form.validate_on_submit():
|
|
user = User.query.filter_by(username=form.username.data).first()
|
|
|
|
if user and user.check_password(form.password.data):
|
|
if not user.active:
|
|
flash('Your account has been disabled. Please contact an administrator.', 'danger')
|
|
return render_template('auth/login.html', form=form)
|
|
if not user.password_set:
|
|
flash(
|
|
'Your account password has not been set yet. '
|
|
'Please check your email for the account setup link.',
|
|
'warning'
|
|
)
|
|
return render_template('auth/login.html', form=form)
|
|
login_user(user, remember=form.remember_me.data)
|
|
# Use validated next URL — never redirect blindly to request.args['next']
|
|
next_page = safe_redirect_url(request.args.get('next'))
|
|
log_action(ACTION_LOGIN, 'User', user.id, user.username)
|
|
flash(f'Welcome back, {user.username}!', 'success')
|
|
return redirect(next_page)
|
|
else:
|
|
# Generic message — don't reveal whether the username exists
|
|
flash('Invalid credentials. Please try again.', 'danger')
|
|
|
|
return render_template('auth/login.html', form=form)
|
|
|
|
|
|
@bp.route('/logout')
|
|
@login_required
|
|
def logout():
|
|
log_action(ACTION_LOGOUT, 'User', current_user.id, current_user.username)
|
|
logout_user()
|
|
flash('Successfully logged out.', 'success')
|
|
return redirect(url_for('auth.login'))
|
|
|
|
|
|
@bp.route('/profile', methods=['GET', 'POST'])
|
|
@login_required
|
|
def profile():
|
|
"""User profile page — view stats and update email/password."""
|
|
from app.models.inspection import Inspection
|
|
from app.models.issue import Issue
|
|
|
|
form = ProfileForm(user=current_user, obj=current_user)
|
|
|
|
if form.validate_on_submit():
|
|
current_user.full_name = form.full_name.data.strip() or None
|
|
current_user.email = form.email.data
|
|
|
|
if form.new_password.data:
|
|
current_user.set_password(form.new_password.data)
|
|
logger.info('AUTH | profile_password_change | user_id=%s username=%s',
|
|
current_user.id, current_user.username)
|
|
|
|
db.session.commit()
|
|
logger.info('AUTH | profile_update | user_id=%s username=%s email=%s',
|
|
current_user.id, current_user.username, current_user.email)
|
|
log_action(ACTION_UPDATE, 'User', current_user.id, current_user.username,
|
|
'self-service profile update')
|
|
flash('Profile updated successfully.', 'success')
|
|
return redirect(url_for('auth.profile'))
|
|
|
|
# ── Activity stats ────────────────────────────────────────────────────
|
|
total_inspections = Inspection.query.filter_by(inspector_id=current_user.id).count()
|
|
completed_inspections = Inspection.query.filter_by(
|
|
inspector_id=current_user.id, status='completed'
|
|
).count()
|
|
|
|
recent_inspections = (
|
|
Inspection.query
|
|
.filter_by(inspector_id=current_user.id)
|
|
.order_by(Inspection.inspection_date.desc())
|
|
.limit(5)
|
|
.all()
|
|
)
|
|
|
|
open_issues = Issue.query.filter_by(
|
|
assigned_to=current_user.id, status='open'
|
|
).count() if hasattr(Issue, 'assigned_to') else 0
|
|
|
|
return render_template(
|
|
'auth/profile.html',
|
|
form=form,
|
|
total_inspections=total_inspections,
|
|
completed_inspections=completed_inspections,
|
|
recent_inspections=recent_inspections,
|
|
open_issues=open_issues,
|
|
)
|
|
|
|
|
|
@bp.route('/users')
|
|
@login_required
|
|
@supervisor_required
|
|
def list_users():
|
|
# Exclude customer accounts — those are managed exclusively via /customers
|
|
users = (
|
|
User.query
|
|
.filter(User.role != 'customer')
|
|
.order_by(User.created_at.desc())
|
|
.all()
|
|
)
|
|
logger.info('AUTH | list_users | admin=%s | internal_users_count=%s',
|
|
current_user.username, len(users))
|
|
return render_template('auth/users.html', users=users)
|
|
|
|
|
|
@bp.route('/users/new', methods=['GET', 'POST'])
|
|
@login_required
|
|
@supervisor_required
|
|
def create_user():
|
|
form = UserForm()
|
|
|
|
if form.validate_on_submit():
|
|
user = User(
|
|
username=form.username.data,
|
|
full_name=form.full_name.data.strip() or None,
|
|
email=form.email.data,
|
|
role=form.role.data
|
|
)
|
|
user.set_password(form.password.data)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
logger.info('AUTH | user_create | admin_id=%s admin=%s new_user=%s role=%s',
|
|
current_user.id, current_user.username, user.username, user.role)
|
|
log_action(ACTION_CREATE, 'User', user.id, user.username,
|
|
f'role={user.role}; email={user.email}')
|
|
flash(f'User {user.username} created successfully.', 'success')
|
|
return redirect(url_for('auth.list_users'))
|
|
|
|
return render_template('auth/user_form.html', form=form, title='Create User')
|
|
|
|
|
|
@bp.route('/users/<int:user_id>/edit', methods=['GET', 'POST'])
|
|
@login_required
|
|
@supervisor_required
|
|
def edit_user(user_id):
|
|
user = db.session.get(User, user_id)
|
|
if user is None:
|
|
abort(404)
|
|
form = UserForm(user=user, obj=user)
|
|
|
|
if form.validate_on_submit():
|
|
user.username = form.username.data
|
|
user.full_name = form.full_name.data.strip() or None
|
|
user.email = form.email.data
|
|
user.role = form.role.data
|
|
|
|
if form.password.data:
|
|
user.set_password(form.password.data)
|
|
|
|
db.session.commit()
|
|
logger.info('AUTH | user_edit | admin_id=%s admin=%s target_user_id=%s target_user=%s',
|
|
current_user.id, current_user.username, user.id, user.username)
|
|
log_action(ACTION_UPDATE, 'User', user.id, user.username,
|
|
f'role={user.role}; email={user.email}')
|
|
flash(f'User {user.username} updated successfully.', 'success')
|
|
return redirect(url_for('auth.list_users'))
|
|
|
|
return render_template('auth/user_form.html', form=form, user=user, title='Edit User')
|
|
|
|
|
|
@bp.route('/users/<int:user_id>/delete', methods=['POST'])
|
|
@login_required
|
|
@supervisor_required
|
|
def delete_user(user_id):
|
|
user = db.session.get(User, user_id)
|
|
if user is None:
|
|
abort(404)
|
|
|
|
if user.id == current_user.id:
|
|
flash('Cannot delete your own account.', 'danger')
|
|
return redirect(url_for('auth.list_users'))
|
|
|
|
# Guard: block deletion if user has related records that would orphan data
|
|
# or violate FK constraints. Issue.assigned_to and IssueComment.user_id carry
|
|
# no ondelete clause, so MySQL defaults to RESTRICT — the DELETE would fail at
|
|
# the DB level without these application-level checks and clear user-facing messages.
|
|
if user.inspections.count() > 0:
|
|
flash(
|
|
f'Cannot delete "{user.username}" — they have existing inspection records. '
|
|
'Deactivate the account instead.',
|
|
'danger'
|
|
)
|
|
return redirect(url_for('auth.list_users'))
|
|
|
|
if user.assigned_issues.count() > 0:
|
|
flash(
|
|
f'Cannot delete "{user.username}" — they have issues assigned to them. '
|
|
'Reassign or resolve those issues first, then deactivate the account.',
|
|
'danger'
|
|
)
|
|
return redirect(url_for('auth.list_users'))
|
|
|
|
from app.models.issue import IssueComment
|
|
if IssueComment.query.filter_by(user_id=user.id).count() > 0:
|
|
flash(
|
|
f'Cannot delete "{user.username}" — they have authored issue comments. '
|
|
'Deactivate the account instead.',
|
|
'danger'
|
|
)
|
|
return redirect(url_for('auth.list_users'))
|
|
|
|
from app.models.inspection import InspectionTemplate
|
|
if InspectionTemplate.query.filter_by(created_by=user.id).count() > 0:
|
|
flash(
|
|
f'Cannot delete "{user.username}" — they have created inspection templates. '
|
|
'Deactivate the account instead.',
|
|
'danger'
|
|
)
|
|
return redirect(url_for('auth.list_users'))
|
|
|
|
username = user.username
|
|
user_id = user.id
|
|
db.session.delete(user)
|
|
db.session.commit()
|
|
logger.info('AUTH | user_delete | admin_id=%s admin=%s deleted_user=%s',
|
|
current_user.id, current_user.username, username)
|
|
log_action(ACTION_DELETE, 'User', user_id, username)
|
|
flash(f'User {username} deleted successfully.', 'success')
|
|
return redirect(url_for('auth.list_users'))
|
|
|
|
@bp.route('/users/<int:user_id>/toggle-active', methods=['POST'])
|
|
@login_required
|
|
@supervisor_required
|
|
def toggle_active(user_id):
|
|
user = db.session.get(User, user_id)
|
|
if user is None:
|
|
abort(404)
|
|
|
|
if user.id == current_user.id:
|
|
flash('You cannot disable your own account.', 'danger')
|
|
return redirect(url_for('auth.list_users'))
|
|
|
|
user.active = not user.active
|
|
db.session.commit()
|
|
|
|
action_label = 'enabled' if user.active else 'disabled'
|
|
logger.info(
|
|
'AUTH | user_%s | admin_id=%s admin=%s target_user=%s',
|
|
action_label, current_user.id, current_user.username, user.username,
|
|
)
|
|
log_action(
|
|
ACTION_UPDATE, 'User', user.id, user.username,
|
|
f'account {action_label} by {current_user.username}',
|
|
)
|
|
flash(f'User {user.username} has been {action_label}.', 'success')
|
|
return redirect(safe_redirect_url(request.referrer, fallback=url_for('auth.list_users')))
|
|
|
|
# ── Notification Matrix ───────────────────────────────────────────────────────
|
|
|
|
@bp.route('/notification-matrix', methods=['GET', 'POST'])
|
|
@login_required
|
|
@admin_required
|
|
def notification_matrix():
|
|
"""Admin-only notification matrix — controls who receives each event type."""
|
|
import json as _json
|
|
from app.models.notification_matrix import (
|
|
NotificationMatrix, MATRIX_EVENTS, MATRIX_ROLES, MATRIX_DEFAULTS,
|
|
)
|
|
|
|
if request.method == 'POST':
|
|
for event_key in MATRIX_EVENTS:
|
|
for role_key, _ in MATRIX_ROLES:
|
|
row = NotificationMatrix.query.filter_by(
|
|
event_type=event_key, role_key=role_key
|
|
).first()
|
|
if row is None:
|
|
row = NotificationMatrix(event_type=event_key, role_key=role_key)
|
|
db.session.add(row)
|
|
|
|
if role_key == 'custom':
|
|
raw = request.form.get(f'custom_{event_key}', '').strip()
|
|
# Parse comma-separated emails into a JSON list
|
|
emails = [e.strip() for e in raw.split(',') if e.strip()]
|
|
row.custom_emails = _json.dumps(emails)
|
|
row.enabled = bool(emails)
|
|
else:
|
|
row.enabled = bool(request.form.get(f'matrix_{event_key}_{role_key}'))
|
|
|
|
db.session.commit()
|
|
log_action(ACTION_UPDATE, 'NotificationMatrix', None,
|
|
'Notification Matrix', 'admin updated notification matrix')
|
|
logger.info('NOTIFICATION MATRIX UPDATED | by=%s', current_user.username)
|
|
flash('Notification matrix saved successfully.', 'success')
|
|
return redirect(url_for('auth.notification_matrix'))
|
|
|
|
# Build current state dict: {event_key: {role_key: enabled/emails}}
|
|
all_rows = NotificationMatrix.query.all()
|
|
state = {} # event_key -> role_key -> row
|
|
for row in all_rows:
|
|
state.setdefault(row.event_type, {})[row.role_key] = row
|
|
|
|
return render_template(
|
|
'auth/notification_matrix.html',
|
|
matrix_events = MATRIX_EVENTS,
|
|
matrix_roles = MATRIX_ROLES,
|
|
defaults = MATRIX_DEFAULTS,
|
|
state = state,
|
|
)
|