1002 lines
42 KiB
Python
1002 lines
42 KiB
Python
from flask import Blueprint, render_template, redirect, url_for, flash, request, abort, session
|
|
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
|
|
from app.utils import mfa
|
|
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
|
|
from app.tenancy.session_binding import bind_session_tenant, SESSION_TENANT_KEY
|
|
|
|
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)
|
|
# ── Two-factor gate (phase35) ────────────────────────────────
|
|
# If this account has TOTP enabled, defer login_user() to the
|
|
# second-factor step. Password is verified; identity is NOT yet
|
|
# established until the code is confirmed at /auth/mfa.
|
|
if user.mfa_enabled and user.mfa_secret:
|
|
bind_session_tenant() # MT-21: tag before identity enters the session
|
|
session['mfa_pending_user_id'] = user.id
|
|
session['mfa_pending_remember'] = bool(form.remember_me.data)
|
|
session['mfa_pending_next'] = safe_redirect_url(request.args.get('next'))
|
|
logger.info('MFA_CHALLENGE | user=%s', user.username)
|
|
return redirect(url_for('auth.mfa_challenge'))
|
|
|
|
login_user(user, remember=form.remember_me.data)
|
|
bind_session_tenant() # MT-21
|
|
# 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')
|
|
logger.warning('LOGIN_FAILED | ip=%s username=%s',
|
|
request.remote_addr, form.username.data)
|
|
|
|
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'))
|
|
|
|
|
|
# ── Two-factor authentication (phase35) ─────────────────────────────────────
|
|
|
|
@bp.route('/mfa', methods=['GET', 'POST'])
|
|
@limiter.limit('10 per minute; 3 per second')
|
|
def mfa_challenge():
|
|
"""Second-factor step during login. Reached only after a correct password
|
|
for an MFA-enabled account (identity is held pending in the session)."""
|
|
uid = session.get('mfa_pending_user_id')
|
|
if not uid:
|
|
return redirect(url_for('auth.login'))
|
|
user = db.session.get(User, uid)
|
|
if user is None or not user.mfa_enabled or not user.active:
|
|
session.pop('mfa_pending_user_id', None)
|
|
return redirect(url_for('auth.login'))
|
|
|
|
if request.method == 'POST':
|
|
code = request.form.get('code', '')
|
|
use_recovery = bool(request.form.get('recovery'))
|
|
verified = False
|
|
via = 'totp'
|
|
|
|
if use_recovery:
|
|
matched, remaining = mfa.check_and_consume_recovery(user.mfa_recovery_codes, code)
|
|
if matched:
|
|
user.mfa_recovery_codes = remaining
|
|
db.session.commit()
|
|
verified = True
|
|
via = 'recovery'
|
|
else:
|
|
verified = mfa.verify_totp(user.mfa_secret, code)
|
|
|
|
if verified:
|
|
remember = session.pop('mfa_pending_remember', False)
|
|
next_page = session.pop('mfa_pending_next', None)
|
|
session.pop('mfa_pending_user_id', None)
|
|
login_user(user, remember=remember)
|
|
bind_session_tenant() # MT-21
|
|
log_action(ACTION_LOGIN, 'User', user.id, user.username, f'2fa via {via}')
|
|
if via == 'recovery':
|
|
remaining_n = len(user.mfa_recovery_codes or [])
|
|
flash(f'Signed in with a recovery code. {remaining_n} recovery '
|
|
f'code(s) remaining.', 'warning')
|
|
else:
|
|
flash(f'Welcome back, {user.username}!', 'success')
|
|
return redirect(safe_redirect_url(next_page))
|
|
|
|
logger.warning('MFA_FAILED | user=%s ip=%s recovery=%s',
|
|
user.username, request.remote_addr, use_recovery)
|
|
flash('Invalid verification code. Please try again.', 'danger')
|
|
|
|
return render_template('auth/mfa_challenge.html')
|
|
|
|
|
|
@bp.route('/mfa/setup', methods=['GET', 'POST'])
|
|
@login_required
|
|
@supervisor_required # admin + director
|
|
def mfa_setup():
|
|
"""Enroll the current account in TOTP two-factor. Opt-in.
|
|
|
|
The candidate secret is held in the session until the user proves they can
|
|
generate a valid code, so a half-finished enrollment never locks anyone out.
|
|
"""
|
|
if current_user.mfa_enabled:
|
|
flash('Two-factor authentication is already enabled on your account.', 'info')
|
|
return redirect(url_for('auth.profile'))
|
|
|
|
if request.method == 'POST':
|
|
secret = session.get('mfa_setup_secret')
|
|
code = request.form.get('code', '')
|
|
if not secret:
|
|
flash('Your setup session expired. Please start again.', 'warning')
|
|
return redirect(url_for('auth.mfa_setup'))
|
|
if mfa.verify_totp(secret, code):
|
|
plaintext, hashed = mfa.generate_recovery_codes()
|
|
current_user.mfa_secret = secret
|
|
current_user.mfa_enabled = True
|
|
current_user.mfa_recovery_codes = hashed
|
|
db.session.commit()
|
|
session.pop('mfa_setup_secret', None)
|
|
log_action(ACTION_UPDATE, 'User', current_user.id, current_user.username,
|
|
'enabled two-factor authentication')
|
|
logger.info('MFA_ENABLED | user=%s', current_user.username)
|
|
# Recovery codes are shown exactly once, right here.
|
|
return render_template('auth/mfa_recovery.html', codes=plaintext)
|
|
flash('That code did not match. Make sure your device clock is correct '
|
|
'and try again.', 'danger')
|
|
|
|
# GET, or a failed POST: (re)present the QR for the pending secret.
|
|
secret = session.get('mfa_setup_secret') or mfa.new_secret()
|
|
session['mfa_setup_secret'] = secret
|
|
uri = mfa.provisioning_uri(secret, current_user.email or current_user.username)
|
|
return render_template('auth/mfa_setup.html', secret=secret, qr_svg=mfa.qr_svg(uri))
|
|
|
|
|
|
@bp.route('/mfa/disable', methods=['POST'])
|
|
@login_required
|
|
def mfa_disable():
|
|
"""Turn off two-factor. Requires a current authenticator code OR the account
|
|
password, so a merely-hijacked session can't silently strip 2FA."""
|
|
if not current_user.mfa_enabled:
|
|
return redirect(url_for('auth.profile'))
|
|
code = request.form.get('code', '')
|
|
pw = request.form.get('password', '')
|
|
if not (mfa.verify_totp(current_user.mfa_secret, code)
|
|
or (pw and current_user.check_password(pw))):
|
|
flash('Enter a valid authenticator code or your password to disable 2FA.', 'danger')
|
|
return redirect(url_for('auth.profile'))
|
|
current_user.mfa_enabled = False
|
|
current_user.mfa_secret = None
|
|
current_user.mfa_recovery_codes = None
|
|
db.session.commit()
|
|
log_action(ACTION_UPDATE, 'User', current_user.id, current_user.username,
|
|
'disabled two-factor authentication')
|
|
logger.info('MFA_DISABLED | user=%s', current_user.username)
|
|
flash('Two-factor authentication has been disabled.', 'success')
|
|
return redirect(url_for('auth.profile'))
|
|
|
|
|
|
@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.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, and scoped to the caller. Records that merely *reference* this
|
|
user are included only as the user's own row — related entities are NOT
|
|
expanded, so an issue this user commented on contributes the comment, not
|
|
the facility details or the other participants. That keeps a subject-access
|
|
request from becoming a data leak about everyone else.
|
|
|
|
Multi-tenant note: this runs against the caller's own tenant DB via the
|
|
normal request routing, so it can only ever see that tenant's data.
|
|
"""
|
|
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.
|
|
|
|
Two outcomes, chosen automatically:
|
|
|
|
* No records that a hard delete would orphan (same guard rails as the admin
|
|
delete_user route) → the account is deleted outright.
|
|
* Otherwise — the common case, since staff usually have inspection or issue
|
|
history that must be kept for business and audit continuity — the account
|
|
is ANONYMIZED in place: name/email/username replaced with a
|
|
non-identifying placeholder, the password hash invalidated so nobody can
|
|
ever log in as it again, and the account deactivated.
|
|
|
|
Historical records reference the user *id*, not the PII, so they survive the
|
|
anonymization unchanged and the audit trail stays intact. This is the
|
|
balance the regulations expect: erase the identity, keep the ledger.
|
|
"""
|
|
import secrets
|
|
from app.models.issue import Issue as _Issue, IssueComment as _IssueComment
|
|
from app.models.inspection import InspectionTemplate as _InspectionTemplate
|
|
|
|
user = current_user
|
|
|
|
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
|
|
# Random hash nobody holds — the account can never be logged into again.
|
|
user.set_password(secrets.token_hex(32))
|
|
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-side accounts — Customer Director AND Customer Inspector
|
|
# are both managed exclusively via /customers (phase51). Using
|
|
# User.CUSTOMER_ROLES rather than != 'customer' is what moves the customer
|
|
# inspectors off this page.
|
|
users = (
|
|
User.query
|
|
.filter(~User.role.in_(User.CUSTOMER_ROLES))
|
|
.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)
|
|
|
|
|
|
def _redirect_if_customer_account(user):
|
|
"""Send customer-side accounts back to Customer Management.
|
|
|
|
phase51 moved Customer Director + Customer Inspector wholly under
|
|
/customers. These accounts are no longer listed here, but the /auth/users
|
|
URLs are still reachable by hand — and editing one through UserForm would
|
|
fail anyway ('external_inspector' is no longer an offered role choice, so
|
|
SelectField would reject the existing value). Redirect instead of 404 so an
|
|
old bookmark lands on the page that now owns the account.
|
|
|
|
Returns a response to return, or None to continue.
|
|
"""
|
|
if user is not None and user.is_customer_account:
|
|
flash(f'{user.display_name} is a {user.role_label} account and is '
|
|
f'managed in Customer Management.', 'info')
|
|
return redirect(url_for('customers.manage', customer_id=user.id))
|
|
return None
|
|
|
|
|
|
@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
|
|
|
|
# phase51 — the invitation branch that used to live here moved to
|
|
# Customer Management along with the Customer Inspector role. Every
|
|
# role this form still offers is OUR OWN staff, created with an
|
|
# admin-set password. Customer-side accounts are invited (they choose
|
|
# their own username and password) via customers.create().
|
|
#
|
|
# 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.
|
|
if not form.password.data:
|
|
flash('Please set a password for the new user.', '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.strip() or None,
|
|
email=form.email.data.strip().lower(),
|
|
role=role,
|
|
password_set=True,
|
|
)
|
|
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/<int:user_id>/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)
|
|
moved = _redirect_if_customer_account(user)
|
|
if moved:
|
|
return moved
|
|
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.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/<int:user_id>/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)
|
|
moved = _redirect_if_customer_account(user)
|
|
if moved:
|
|
return moved
|
|
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/<int:user_id>/assign-contracts', methods=['GET', 'POST'])
|
|
@login_required
|
|
@admin_required
|
|
def assign_inspector_contracts(user_id):
|
|
user = db.session.get(User, user_id)
|
|
# MT-15: external inspectors are scoped by the same InspectorAssignment
|
|
# rows, so this page must accept them too.
|
|
if user is None or not user.is_inspector:
|
|
abort(404)
|
|
# A Customer Inspector is scoped by exactly these rows, but the page that
|
|
# owns them is now customers.manage — one editor per account, not two.
|
|
moved = _redirect_if_customer_account(user)
|
|
if moved:
|
|
return moved
|
|
|
|
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/<int:user_id>/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/<int:user_id>/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)}'
|
|
|
|
# Branded From as a (display_name, address) tuple, exactly as
|
|
# customers._send_invite_email does. The display NAME tracks the tenant; the
|
|
# ADDRESS is branded only for DNS-authorized domains and otherwise stays the
|
|
# authenticated SMTP identity so the mail still delivers. The previous
|
|
# `noreply@{host}` sent from whatever host the browser was on, which is not
|
|
# an authorized sender for that domain — the mail server accepted it and it
|
|
# was then dropped downstream by SPF/DMARC. See app/utils/mail_utils.py and
|
|
# CLAUDE.md rule 64.
|
|
from app.utils.mail_utils import branded_sender
|
|
sender = branded_sender(effective_base)
|
|
|
|
html_body = render_template_string("""<!DOCTYPE html>
|
|
<html>
|
|
<body style="font-family:Arial,sans-serif;color:#333;max-width:600px;margin:auto;">
|
|
<h2 style="color:#0d6efd;">Password Reset Request</h2>
|
|
<p>Hi {{ name }},</p>
|
|
<p>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
|
|
<strong>1 hour</strong>.</p>
|
|
<p>
|
|
<a href="{{ link }}"
|
|
style="background:#0d6efd;color:#fff;padding:12px 24px;
|
|
text-decoration:none;border-radius:4px;display:inline-block;font-weight:bold;">
|
|
Reset My Password
|
|
</a>
|
|
</p>
|
|
<p style="font-size:13px;color:#666;">
|
|
If you did not request a password reset, you can safely ignore this email.
|
|
Your password will not change.
|
|
</p>
|
|
<p style="font-size:13px;color:#888;">
|
|
Or copy this URL:<br>
|
|
<a href="{{ link }}" style="color:#0d6efd;">{{ link }}</a>
|
|
</p>
|
|
<hr style="border:none;border-top:1px solid #eee;margin-top:32px;">
|
|
<p style="font-size:12px;color:#888;">Janitorial QC System — do not reply.</p>
|
|
</body>
|
|
</html>""", 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():
|
|
# Explicit case-insensitive lookup. This is NOT fixing a live bug: no
|
|
# table here declares a COLLATE, so `users.email` inherits the utf8mb4
|
|
# default (utf8mb4_general_ci / utf8mb4_0900_ai_ci), both of which are
|
|
# case-insensitive — a bare `== lower(input)` already matched a
|
|
# mixed-case stored address. The point is to stop depending on that
|
|
# server default: under a binary/_bin collation the bare comparison
|
|
# would silently find nothing and still show the success message below.
|
|
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/<token>', 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=<signed_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
|
|
|
|
# ── MT-21: re-tag the session for the impersonated tenant ────────────────
|
|
# Identity in the session is tenant-tagged (User.get_id), and from the next
|
|
# request onward the middleware binds the impersonated tenant's database.
|
|
# Without re-tagging, load_user() would correctly reject the tag issued by
|
|
# the host tenant and the superadmin would land on a login page.
|
|
#
|
|
# This preserves the pre-existing impersonation semantics EXACTLY: the
|
|
# numeric user id carries over, so the superadmin is loaded as the
|
|
# same-numbered user in the target tenant's database. That behaviour is
|
|
# arbitrary and worth revisiting (see MULTI_TENANT_PLAN.md open items) —
|
|
# but changing it is a separate decision, not a security fix.
|
|
flask_session[SESSION_TENANT_KEY] = tenant_id
|
|
raw_uid = flask_session.get('_user_id')
|
|
if raw_uid is not None:
|
|
numeric_uid = str(raw_uid).rpartition(':')[2]
|
|
flask_session['_user_id'] = f'{tenant_id}:{numeric_uid}'
|
|
|
|
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'<a href="{end_url}" class="alert-link">End impersonation</a>'
|
|
),
|
|
'warning',
|
|
)
|
|
return redirect(url_for('dashboard.index'))
|
|
|
|
|
|
@bp.route('/impersonate/end')
|
|
@login_required
|
|
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)
|
|
# MT-21: the session identity is still tagged for the impersonated tenant.
|
|
# Drop it rather than carrying it back to the superadmin's own host, where
|
|
# the middleware would clear it on tenant mismatch anyway.
|
|
logout_user()
|
|
flask_session.clear()
|
|
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) |