from flask import Blueprint, render_template, redirect, url_for, flash, request, abort from urllib.parse import urlparse 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, ForgotPasswordForm, ResetPasswordForm 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 from app.tenancy.gates import quota_soft_check 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 @admin_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() ) # Build a map of inspector_id -> assignment count for the Contracts column from app.models.inspector_assignment import InspectorAssignment from sqlalchemy import func rows = ( db.session.query( InspectorAssignment.user_id, func.count(InspectorAssignment.id).label('cnt'), ) .group_by(InspectorAssignment.user_id) .all() ) inspector_contract_counts = {r.user_id: r.cnt for r in rows} logger.info('AUTH | list_users | admin=%s | internal_users_count=%s', current_user.username, len(users)) return render_template('auth/users.html', users=users, inspector_contract_counts=inspector_contract_counts) @bp.route('/users/new', methods=['GET', 'POST']) @login_required @admin_required @quota_soft_check('users') def create_user(): form = UserForm() # Directors may not assign roles — new users created by a director default # to inspector. Only admins may set an arbitrary role at creation time. director_editing = current_user.role == 'director' if form.validate_on_submit(): role = 'inspector' if director_editing else form.role.data user = User( username=form.username.data, full_name=form.full_name.data.strip() or None, email=form.email.data, role=role ) 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', director_editing=director_editing) @bp.route('/users//edit', methods=['GET', 'POST']) @login_required @admin_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) # Directors may not change another user's role — that privilege is admin-only. # The role field is removed from the form for directors so it cannot be # submitted at all, and the existing role value is preserved on save. director_editing = current_user.role == 'director' 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 if not director_editing: 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', director_editing=director_editing) @bp.route('/users//assign-contracts', methods=['GET', 'POST']) @login_required @admin_required def assign_inspector_contracts(user_id): user = db.session.get(User, user_id) if user is None or user.role != 'inspector': abort(404) from app.models.project import Project from app.models.inspector_assignment import InspectorAssignment from app.utils.time_utils import now_eastern projects = Project.query.filter_by(active=True).order_by(Project.name).all() if request.method == 'POST': selected_ids = set(request.form.getlist('project_ids', type=int)) existing = InspectorAssignment.query.filter_by(user_id=user_id).all() existing_pids = {a.project_id for a in existing} for a in existing: if a.project_id not in selected_ids: db.session.delete(a) for pid in selected_ids: if pid not in existing_pids: db.session.add(InspectorAssignment( user_id = user_id, project_id = pid, created_at = now_eastern(), )) db.session.commit() log_action(ACTION_UPDATE, 'User', user.id, user.username, f'inspector_assignments={sorted(selected_ids)}') flash(f'Contract assignments updated for {user.display_name}.', 'success') return redirect(url_for('auth.list_users')) assigned_pids = { a.project_id for a in InspectorAssignment.query.filter_by(user_id=user_id).all() } return render_template('auth/inspector_assignments.html', user=user, projects=projects, assigned_pids=assigned_pids) @bp.route('/users//delete', methods=['POST']) @login_required @admin_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//toggle-active', methods=['POST']) @login_required @admin_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, ) # ── Forgot / Reset Password (public) ───────────────────────────────────────── def _send_password_reset_email(user, token, base_url=None): """Send a password-reset link email. Mirrors _send_invite_email in customers.py.""" from flask import current_app, render_template_string, url_for as _url_for from flask_mail import Message from app import mail from urllib.parse import urlparse import threading if not current_app.config.get('MAIL_SERVER'): logger.warning('RESET EMAIL SKIPPED | no MAIL_SERVER | user=%s', user.username) return effective_base = (base_url or current_app.config.get('APP_BASE_URL', '')).rstrip('/') reset_link = f'{effective_base}{_url_for("auth.reset_password", token=token)}' host = urlparse(effective_base).netloc or 'janitorialqc.local' sender = f'noreply@{host}' html_body = render_template_string("""

Password Reset Request

Hi {{ name }},

We received a request to reset your password for the Janitorial QC portal. Click the button below to choose a new password. This link expires in 1 hour.

Reset My Password

If you did not request a password reset, you can safely ignore this email. Your password will not change.

Or copy this URL:
{{ link }}

Janitorial QC System — do not reply.

""", name=user.display_name, link=reset_link) text_body = ( f'Hi {user.display_name},\n\n' f'We received a request to reset your JQC password.\n' f'Click the link below to reset it (expires in 1 hour):\n\n{reset_link}\n\n' f'If you did not request this, ignore this email.\n\nJanitorial QC System' ) msg = Message( subject = '[JQC] Password reset request', sender = sender, recipients = [user.email], body = text_body, html = html_body, ) app = current_app._get_current_object() def _send(): with app.app_context(): try: mail.send(msg) logger.info('RESET EMAIL SENT | to=%s | user=%s', user.email, user.username) except Exception as exc: logger.error('RESET EMAIL FAILED | to=%s | error=%s', user.email, exc) threading.Thread(target=_send, daemon=True).start() @bp.route('/forgot-password', methods=['GET', 'POST']) @limiter.limit('10 per hour') def forgot_password(): if current_user.is_authenticated: return redirect(url_for('dashboard.index')) form = ForgotPasswordForm() if form.validate_on_submit(): user = User.query.filter_by(email=form.email.data.strip().lower()).first() if user and user.active: token = user.generate_set_password_token(expires_hours=1) db.session.commit() _send_password_reset_email(user, token, base_url=request.host_url) logger.info('AUTH | forgot_password | user=%s | email=%s', user.username, user.email) # Always show the same message — never reveal whether the email exists flash( 'If an account with that email address exists, a password reset link ' 'has been sent. Please check your inbox (and spam folder).', 'info' ) return redirect(url_for('auth.login')) return render_template('auth/forgot_password.html', form=form) @bp.route('/reset-password/', methods=['GET', 'POST']) def reset_password(token): if current_user.is_authenticated: return redirect(url_for('dashboard.index')) user = User.verify_set_password_token(token) if user is None: flash('This password reset link is invalid or has expired.', 'danger') return redirect(url_for('auth.forgot_password')) form = ResetPasswordForm() if form.validate_on_submit(): user.set_password(form.password.data) user.clear_set_password_token() db.session.commit() log_action(ACTION_UPDATE, 'User', user.id, user.username, 'password reset via forgot-password link') flash('Your password has been reset successfully. Please log in.', 'success') return redirect(url_for('auth.login')) return render_template('auth/reset_password.html', form=form, user=user) # ── MT-4: Superadmin impersonation ──────────────────────────────────────────── @bp.route('/impersonate') def impersonate_entry(): """ Validate a superadmin impersonation token and bind the session to a tenant. Called by the control panel redirect: GET /auth/impersonate?token= Sets session['impersonating_tenant_id'] which the tenancy middleware reads to short-circuit normal Host resolution for the duration of the session. This route is CSRF-exempt by nature — the HMAC token already provides auth. """ from flask import session as flask_session token = request.args.get('token', '') if not token: flash('Missing impersonation token.', 'danger') return redirect(url_for('auth.login')) try: from control.panel.impersonate import validate_token payload = validate_token(token) except ValueError as e: logger.warning('AUTH | impersonate_invalid | reason=%s', e) flash('Invalid or expired impersonation link.', 'danger') return redirect(url_for('auth.login')) tenant_id = payload.get('tid') superadmin_id = payload.get('said') flask_session['impersonating_tenant_id'] = tenant_id flask_session['impersonating_superadmin_id'] = superadmin_id logger.info('AUTH | impersonate_start | sa=%s tenant=%s', superadmin_id, tenant_id) import os from markupsafe import Markup end_url = url_for('auth.impersonate_end') flash( Markup( f'Impersonating tenant #{tenant_id} as superadmin. ' f'End impersonation' ), 'warning', ) return redirect(url_for('dashboard.index')) @bp.route('/impersonate/end') def impersonate_end(): """Clear impersonation session keys and redirect back to the control panel.""" from flask import session as flask_session import os flask_session.pop('impersonating_tenant_id', None) flask_session.pop('impersonating_superadmin_id', None) panel_url = f"https://admin.{os.environ.get('TENANT_BASE_DOMAIN', 'jqc.app')}" logger.info('AUTH | impersonate_end | redirecting to panel') return redirect(panel_url)