nngo/JQC_multi_tenant
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

First commit

Browse Source
This commit is contained in:
nngo
2026-06-26 09:04:34 -04:00
commit 77678ed724
166 changed files with 34842 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app/routes/auth.py
+519
View File
@@ -0,0 +1,519 @@
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
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
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/<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)
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/<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)
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/<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
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("""<!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():
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/<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)