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') @login_required 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. Requires an authenticated user so the HMAC token alone cannot grant access to an anonymous session. """ 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') if tenant_id is None: logger.warning('AUTH | impersonate_invalid | reason=missing tid in payload') flash('Invalid impersonation token (missing tenant).', 'danger') return redirect(url_for('auth.login')) 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)