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, ForgotPasswordForm, ResetPasswordForm from app.utils.decorators import admin_required, supervisor_required, safe_redirect_url import logging import secrets 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 or '').strip() or None current_user.email = form.email.data.strip().lower() 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, ) # ── Self-service data export (GDPR Art. 15/20, CCPA right-to-know) ──────────── @bp.route('/my-data/export') @login_required def export_my_data(): """Download a JSON snapshot of everything this account's own records hold: profile fields, inspections performed, issues reported/assigned/commented on, and the audit log entries recorded against this user id. Read-only — does not touch other users' data even where it references this user (e.g. an issue this user commented on is included, but the facility/other participants' details are not expanded).""" from flask import Response import json from app.models.inspection import Inspection from app.models.issue import Issue, IssueComment from app.models.audit import AuditLog from app.utils.time_utils import now_eastern user = current_user payload = { 'exported_at': now_eastern().isoformat(), 'profile': { 'id': user.id, 'username': user.username, 'full_name': user.full_name, 'email': user.email, 'role': user.role, 'created_at': user.created_at.isoformat() if user.created_at else None, 'active': user.active, }, 'inspections_performed': [ { 'id': i.id, 'facility_id': i.facility_id, 'inspection_date': i.inspection_date.isoformat() if i.inspection_date else None, 'overall_score': i.overall_score, 'status': i.status, } for i in Inspection.query.filter_by(inspector_id=user.id).all() ], 'issues_reported': [ {'id': iss.id, 'facility_id': iss.facility_id, 'description': iss.description, 'status': iss.status, 'severity': iss.severity, 'reported_at': iss.reported_at.isoformat() if iss.reported_at else None} for iss in Issue.query.filter_by(reported_by=user.id).all() ], 'issues_assigned': [ {'id': iss.id, 'facility_id': iss.facility_id, 'description': iss.description, 'status': iss.status, 'severity': iss.severity} for iss in Issue.query.filter_by(assigned_to=user.id).all() ], 'issue_comments_authored': [ {'id': c.id, 'issue_id': c.issue_id, 'body': c.body, 'created_at': c.created_at.isoformat() if c.created_at else None} for c in IssueComment.query.filter_by(user_id=user.id).all() ], 'audit_log_entries': [ {'id': a.id, 'action': a.action, 'entity_type': a.entity_type, 'entity_id': a.entity_id, 'entity_label': a.entity_label, 'created_at': a.created_at.isoformat() if a.created_at else None} for a in AuditLog.query.filter_by(user_id=user.id).all() ], } log_action(ACTION_UPDATE, 'User', user.id, user.username, 'self-service data export') logger.info('AUTH | export_my_data | user_id=%s username=%s', user.id, user.username) body = json.dumps(payload, indent=2, default=str) return Response( body, mimetype='application/json', headers={'Content-Disposition': f'attachment; filename=jqc_my_data_{user.id}.json'}, ) # ── Self-service erasure request (GDPR Art. 17, CCPA right-to-delete) ───────── @bp.route('/my-data/delete-request', methods=['POST']) @login_required def request_my_data_deletion(): """Erase this account's PII on request. If the account has no records that would be orphaned by a hard delete (same guard rails as the admin delete_user route), it is deleted outright. Otherwise — the common case, since inspectors/staff usually have inspection or issue history that must be kept for business/audit continuity — the account is anonymized in place: name/email/username are replaced with a non-identifying placeholder, the password hash is invalidated, and the account is deactivated. Historical records (which reference the user id, not the PII) are preserved unchanged.""" user = current_user from app.models.issue import Issue as _Issue, IssueComment as _IssueComment from app.models.inspection import InspectionTemplate as _InspectionTemplate blocking = ( user.inspections.count() > 0 or _Issue.query.filter_by(assigned_to=user.id).count() > 0 or _IssueComment.query.filter_by(user_id=user.id).count() > 0 or _InspectionTemplate.query.filter_by(created_by=user.id).count() > 0 ) username = user.username user_id = user.id if not blocking: db.session.delete(user) db.session.commit() logout_user() logger.info('AUTH | self_delete | user_id=%s username=%s', user_id, username) log_action(ACTION_DELETE, 'User', user_id, username, 'self-service account deletion') flash('Your account and data have been permanently deleted.', 'success') return redirect(url_for('auth.login')) placeholder = f'deleted_user_{user_id}' user.full_name = None user.email = f'{placeholder}@deleted.local' user.username = placeholder user.set_password(secrets.token_hex(32)) # invalidate — no one can log in as this account again user.active = False db.session.commit() logger.info('AUTH | self_anonymize | user_id=%s (had blocking records, hard delete not possible)', user_id) log_action(ACTION_UPDATE, 'User', user_id, placeholder, 'self-service erasure request — anonymized (blocking records retained for audit/business continuity)') logout_user() flash('Your personal information has been removed and your account deactivated. ' 'Historical records tied to your account id are retained for audit continuity but no longer identify you.', 'success') return redirect(url_for('auth.login')) @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 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 # phase51 — an external inspector works for the customer or a third # party, so we never set a password on their behalf. They are invited # exactly like a customer: created with password_set=False (which the # login route refuses until they finish), given a one-time token, and # emailed a link to choose their own username and password. invite = (role == 'external_inspector') # UserForm.password is Optional() because the same form is used for # EDIT, where blank means "keep current". On CREATE a blank password # would otherwise store the hash of an empty string, so require one # unless the account is being invited to choose their own. if not invite and not form.password.data: flash('Please set a password, or choose the External Inspector role ' 'to send an invitation instead.', 'danger') return render_template('auth/user_form.html', form=form, user=None, title='Create User', director_editing=director_editing) user = User( username=form.username.data, full_name=(form.full_name.data or '').strip() or None, email=form.email.data.strip().lower(), role=role, password_set=not invite, ) if invite: # A random unguessable placeholder — password_set=False already # blocks login, but never leave an account holding a known or # empty-string hash. user.set_password(secrets.token_hex(32)) else: user.set_password(form.password.data) db.session.add(user) db.session.flush() # need user.id before minting the token token = user.generate_set_password_token(expires_hours=72) if invite else None db.session.commit() logger.info('AUTH | user_create | admin_id=%s admin=%s new_user=%s role=%s invite=%s', current_user.id, current_user.username, user.username, user.role, invite) log_action(ACTION_CREATE, 'User', user.id, user.username, f'role={user.role}; email={user.email}; invite_sent={invite}') if invite: # Reuses the customer invitation email — the copy ("an account has # been created for you… set your password") is already correct for # any invited account. Imported inside the function to keep the # auth ↔ customers import graph acyclic. from app.routes.customers import _send_invite_email _send_invite_email(user, token, base_url=request.host_url) flash(f'External inspector {user.display_name} created. An invitation ' f'email has been sent to {user.email} with a link to set their ' f'username and password.', 'success') else: 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 or '').strip() or None user.email = form.email.data.strip().lower() 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//resend-invite', methods=['POST']) @login_required @admin_required def resend_invite(user_id): """Re-send the set-password invitation for an account still awaiting setup. Without this an invitation that bounces, is deleted or expires leaves the account permanently unusable — password_set=False blocks login and only a valid token can clear it. Mirrors customers.resend_invite for staff-side accounts (currently only external inspectors are ever invited this way). """ user = db.session.get(User, user_id) if user is None: abort(404) if user.password_set: flash(f'{user.display_name} has already completed their account setup.', 'info') return redirect(url_for('auth.list_users')) # A fresh token invalidates the previous link. token = user.generate_set_password_token(expires_hours=72) db.session.commit() logger.info('AUTH | resend_invite | admin=%s user=%s', current_user.username, user.username) log_action(ACTION_UPDATE, 'User', user.id, user.username, 'invitation email resent') from app.routes.customers import _send_invite_email _send_invite_email(user, token, base_url=request.host_url) flash(f'Invitation resent to {user.email}.', 'success') return redirect(url_for('auth.list_users')) @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 not user.is_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 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)}' # From MUST be the authenticated SMTP identity, otherwise the mail server # accepts the message but it is dropped downstream by SPF/DMARC/relay policy # (a per-host noreply@ is NOT an authorized sender). The per-domain # host is still reflected in the reset LINK above, preserving rule 64's # multi-domain intent without breaking deliverability. sender = (current_app.config.get('MAIL_DEFAULT_SENDER') or current_app.config.get('MAIL_USERNAME') or 'noreply@janitorialqc.local') 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(): # Case-insensitive lookup: emails are stored with inconsistent casing # across create/edit/import paths, so a plain lowercased == match can # silently miss a mixed-case stored address and send nothing. email_input = form.email.data.strip().lower() user = User.query.filter( db.func.lower(User.email) == email_input ).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 | reset link dispatched | user=%s | email=%s', user.username, user.email) else: # No leak to the user (generic message below), but log for diagnosis. logger.info('AUTH | forgot_password | no active account for email=%s', email_input) # 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)