Files
IT_Ticket_System/app/routes/admin.py
T
2026-03-26 17:34:10 -04:00

540 lines
24 KiB
Python

import logging
import os
import uuid
from datetime import datetime, timedelta
from functools import wraps
from flask import Blueprint, render_template, redirect, url_for, flash, request, abort, jsonify, current_app, send_from_directory
from flask_login import login_required, current_user
from werkzeug.utils import secure_filename
import bleach
from app import db
from app.models import (User, Ticket, Comment, ActivityLog, KnowledgeBase,
KBAttachment, UserRole, TicketStatus)
from app.services.log_service import log_action
admin_bp = Blueprint('admin', __name__, url_prefix='/admin')
logger = logging.getLogger(__name__)
# ── KB body HTML sanitisation ─────────────────────────────────────────────────
# TinyMCE produces rich HTML which must be sanitised server-side before
# persistence to prevent stored XSS attacks. Only tags and attributes that
# are safe to render are whitelisted; everything else is stripped.
_KB_ALLOWED_TAGS = set(bleach.sanitizer.ALLOWED_TAGS) | {
'p', 'h1', 'h2', 'h3', 'h4', 'h5', 'h6',
'pre', 'code', 'blockquote', 'hr', 'br',
'table', 'thead', 'tbody', 'tr', 'th', 'td',
'ul', 'ol', 'li', 'dl', 'dt', 'dd',
'img', 'figure', 'figcaption',
'div', 'span', 'section',
'strong', 'em', 'u', 's', 'sub', 'sup',
}
_KB_ALLOWED_ATTRS = {
'*' : ['class', 'id', 'style'],
'a' : ['href', 'title', 'target', 'rel'],
'img': ['src', 'alt', 'width', 'height', 'title'],
'td' : ['colspan', 'rowspan'],
'th' : ['colspan', 'rowspan'],
'col': ['span'],
}
def _sanitize_kb_body(raw_html):
"""Strip disallowed tags/attributes from a TinyMCE-produced HTML body."""
cleaned = bleach.clean(
raw_html or '',
tags = _KB_ALLOWED_TAGS,
attributes = _KB_ALLOWED_ATTRS,
strip = True,
)
logger.debug(f'[KB SANITIZE] input_len={len(raw_html or "")} output_len={len(cleaned)}')
return cleaned
def admin_required(f):
@wraps(f)
def decorated(*args, **kwargs):
if not current_user.is_authenticated or not current_user.is_admin:
abort(403)
return f(*args, **kwargs)
return decorated
def it_required(f):
@wraps(f)
def decorated(*args, **kwargs):
if not current_user.is_authenticated or not current_user.is_it_staff:
abort(403)
return f(*args, **kwargs)
return decorated
# ─── Admin Dashboard ──────────────────────────────────────────────────────────
@admin_bp.route('/')
@login_required
@it_required
def index():
stats = {
'total_tickets' : Ticket.query.count(),
'open' : Ticket.query.filter_by(status=TicketStatus.OPEN).count(),
'in_progress' : Ticket.query.filter_by(status=TicketStatus.IN_PROGRESS).count(),
'resolved' : Ticket.query.filter_by(status=TicketStatus.RESOLVED).count(),
'closed' : Ticket.query.filter_by(status=TicketStatus.CLOSED).count(),
'total_users' : User.query.filter_by(is_active=True).count(),
'employees' : User.query.filter_by(role=UserRole.EMPLOYEE, is_active=True).count(),
'it_staff' : User.query.filter(
User.role.in_([UserRole.IT_STAFF, UserRole.ADMIN]),
User.is_active == True).count(),
}
recent_logs = ActivityLog.query.order_by(ActivityLog.created_at.desc()).limit(20).all()
return render_template('admin/index.html', stats=stats, recent_logs=recent_logs)
# ─── User Management ─────────────────────────────────────────────────────────
@admin_bp.route('/users')
@login_required
@admin_required
def users():
all_users = User.query.order_by(User.created_at.desc()).all()
return render_template('admin/users.html', users=all_users)
@admin_bp.route('/users/new', methods=['GET', 'POST'])
@login_required
@admin_required
def create_user():
if request.method == 'POST':
email = request.form.get('email', '').strip().lower()
username = request.form.get('username', '').strip()
full_name = request.form.get('full_name', '').strip()
department = request.form.get('department', '').strip()
phone = request.form.get('phone', '').strip()
role = request.form.get('role', UserRole.EMPLOYEE)
password = request.form.get('password', '')
confirm = request.form.get('confirm_password', '')
is_active = bool(request.form.get('is_active'))
# ── Validation ────────────────────────────────────────────────────────
if not email or not username or not full_name or not password:
flash('Email, username, full name and password are all required.', 'danger')
return render_template('admin/create_user.html', roles=_roles(), form=request.form)
if User.query.filter_by(email=email).first():
flash('That email address is already registered.', 'danger')
return render_template('admin/create_user.html', roles=_roles(), form=request.form)
if User.query.filter_by(username=username).first():
flash('That username is already taken.', 'danger')
return render_template('admin/create_user.html', roles=_roles(), form=request.form)
if password != confirm:
flash('Passwords do not match.', 'danger')
return render_template('admin/create_user.html', roles=_roles(), form=request.form)
if len(password) < 8:
flash('Password must be at least 8 characters.', 'danger')
return render_template('admin/create_user.html', roles=_roles(), form=request.form)
# ── Create ────────────────────────────────────────────────────────────
user = User(
email = email,
username = username,
full_name = full_name,
department = department,
phone = phone,
role = role,
is_active = is_active,
)
user.set_password(password)
db.session.add(user)
log_action(current_user.id, 'admin_user_create', 'user', user.id,
f'email={email} role={role}')
db.session.commit()
logger.info(f'[ADMIN USER CREATE] user_id={user.id} email={email} role={role} by admin_id={current_user.id}')
flash(f'User {full_name} ({email}) created successfully.', 'success')
return redirect(url_for('admin.users'))
return render_template('admin/create_user.html', roles=_roles(), form={})
@admin_bp.route('/users/<int:user_id>/edit', methods=['GET', 'POST'])
@login_required
@admin_required
def edit_user(user_id):
user = User.query.get_or_404(user_id)
if request.method == 'POST':
old_role = user.role
user.full_name = request.form.get('full_name', user.full_name).strip()
user.department= request.form.get('department', user.department).strip()
user.phone = request.form.get('phone', user.phone or '').strip()
user.role = request.form.get('role', user.role)
user.is_active = bool(request.form.get('is_active'))
new_pw = request.form.get('new_password', '')
if new_pw:
user.set_password(new_pw)
logger.info(f'[ADMIN PASSWORD RESET] target_user_id={user.id} by admin_id={current_user.id}')
log_action(current_user.id, 'admin_user_edit', 'user', user.id,
f'role_change={old_role}->{user.role} active={user.is_active}')
db.session.commit()
logger.info(f'[ADMIN USER EDIT] user_id={user.id} by admin_id={current_user.id}')
flash('User updated.', 'success')
return redirect(url_for('admin.users'))
return render_template('admin/edit_user.html', user=user, roles=_roles())
@admin_bp.route('/users/<int:user_id>/delete', methods=['POST'])
@login_required
@admin_required
def delete_user(user_id):
user = User.query.get_or_404(user_id)
if user.id == current_user.id:
flash('You cannot delete your own account.', 'danger')
return redirect(url_for('admin.users'))
user.is_active = False
log_action(current_user.id, 'admin_user_deactivate', 'user', user.id)
db.session.commit()
logger.info(f'[ADMIN USER DEACTIVATE] user_id={user.id} by admin_id={current_user.id}')
flash('User deactivated.', 'success')
return redirect(url_for('admin.users'))
# ─── Ticket Management (IT) ───────────────────────────────────────────────────
@admin_bp.route('/tickets')
@login_required
@it_required
def all_tickets():
page = request.args.get('page', 1, type=int)
status = request.args.get('status', '')
priority = request.args.get('priority', '')
assigned = request.args.get('assigned', '')
q = Ticket.query
if status: q = q.filter_by(status=status)
if priority: q = q.filter_by(priority=priority)
if assigned == 'me': q = q.filter_by(assigned_to_id=current_user.id)
elif assigned == 'unassigned': q = q.filter_by(assigned_to_id=None)
tickets = q.order_by(Ticket.created_at.desc()).paginate(page=page, per_page=25)
return render_template('admin/tickets.html', tickets=tickets,
status=status, priority=priority, assigned=assigned)
# ─── Knowledge Base Management ────────────────────────────────────────────────
@admin_bp.route('/kb')
@login_required
@it_required
def kb_list():
q = request.args.get('q', '').strip()
category = request.args.get('category', '')
author_id = request.args.get('author_id', '', type=str)
published = request.args.get('published', '') # 'yes' | 'no' | ''
query = KnowledgeBase.query
if q:
query = query.filter(
KnowledgeBase.title.ilike(f'%{q}%') |
KnowledgeBase.tags.ilike(f'%{q}%')
)
if category:
query = query.filter_by(category=category)
if author_id:
query = query.filter_by(author_id=int(author_id))
if published == 'yes':
query = query.filter_by(is_published=True)
elif published == 'no':
query = query.filter_by(is_published=False)
articles = query.order_by(KnowledgeBase.created_at.desc()).all()
# Build filter option lists from existing data
categories = sorted({a.category for a in KnowledgeBase.query.with_entities(KnowledgeBase.category).distinct() if a.category})
authors = User.query.filter(
User.id.in_(
db.session.query(KnowledgeBase.author_id).distinct()
),
User.is_active == True,
).order_by(User.full_name).all()
return render_template('admin/kb_list.html',
articles = articles,
categories = categories,
authors = authors,
q = q,
sel_category = category,
sel_author_id = author_id,
sel_published = published,
)
# ── Helpers ──────────────────────────────────────────────────────────────────
_KB_ALLOWED_EXT = {
'png', 'jpg', 'jpeg', 'gif', 'webp', # images
'pdf', 'doc', 'docx', 'xls', 'xlsx', # documents
'ppt', 'pptx', 'txt', 'csv', 'zip', 'log', # misc
}
def _kb_allowed(filename):
return '.' in filename and filename.rsplit('.', 1)[1].lower() in _KB_ALLOWED_EXT
def _save_kb_file(file, article_id):
"""Save an uploaded file and return a KBAttachment (not yet committed)."""
filename = secure_filename(file.filename)
ext = filename.rsplit('.', 1)[1].lower() if '.' in filename else 'bin'
stored_name = f"{uuid.uuid4().hex}.{ext}"
upload_dir = current_app.config['UPLOAD_FOLDER']
os.makedirs(upload_dir, exist_ok=True)
filepath = os.path.join(upload_dir, stored_name)
file.save(filepath)
return KBAttachment(
article_id = article_id,
filename = filename,
stored_name = stored_name,
mime_type = file.content_type,
file_size = os.path.getsize(filepath),
uploaded_by = current_user.id,
)
# ── TinyMCE image upload endpoint ─────────────────────────────────────────────
@admin_bp.route('/kb/upload-image', methods=['POST'])
@login_required
@it_required
def kb_upload_image():
"""
TinyMCE image upload handler.
Returns JSON: {"location": "<url>"} on success.
"""
f = request.files.get('file')
if not f or not f.filename:
return jsonify({'error': 'No file provided'}), 400
ext = f.filename.rsplit('.', 1)[-1].lower() if '.' in f.filename else ''
if ext not in {'png', 'jpg', 'jpeg', 'gif', 'webp'}:
return jsonify({'error': 'Only image files are accepted (PNG, JPG, GIF, WEBP)'}), 400
stored_name = f"{uuid.uuid4().hex}.{ext}"
upload_dir = current_app.config['UPLOAD_FOLDER']
os.makedirs(upload_dir, exist_ok=True)
f.save(os.path.join(upload_dir, stored_name))
logger.info(f'[KB IMAGE UPLOAD] stored_name={stored_name} by user_id={current_user.id}')
location = url_for('admin.kb_serve_file', stored_name=stored_name)
return jsonify({'location': location})
# ── File-serve route (images embedded in articles + attachment downloads) ─────
@admin_bp.route('/kb/files/<path:stored_name>')
@login_required
def kb_serve_file(stored_name):
"""Serve a KB attachment file. Login required — no public access."""
upload_dir = current_app.config['UPLOAD_FOLDER']
return send_from_directory(upload_dir, stored_name)
# ── Delete a single KB attachment ─────────────────────────────────────────────
@admin_bp.route('/kb/<int:article_id>/attachments/<int:att_id>/delete', methods=['POST'])
@login_required
@it_required
def kb_delete_attachment(article_id, att_id):
att = KBAttachment.query.filter_by(id=att_id, article_id=article_id).first_or_404()
upload_dir = current_app.config['UPLOAD_FOLDER']
filepath = os.path.join(upload_dir, att.stored_name)
if os.path.exists(filepath):
os.remove(filepath)
log_action(current_user.id, 'kb_attachment_delete', 'kb_attachment', att.id,
f'article_id={article_id} filename={att.filename}')
logger.info(f'[KB ATTACHMENT DELETE] att_id={att.id} article_id={article_id} by user_id={current_user.id}')
db.session.delete(att)
db.session.commit()
logger.info(f'[KB ATTACHMENT DELETE] att_id={att.id} article_id={article_id} completed')
# Return JSON so the edit page can remove the row without a full reload
return jsonify({'ok': True, 'att_id': att.id})
@admin_bp.route('/kb/<int:article_id>/attachments', methods=['GET'])
@login_required
@it_required
def kb_get_attachments(article_id):
"""Return current attachment list as JSON for dynamic UI refresh."""
from app.models import KBAttachment
atts = KBAttachment.query.filter_by(article_id=article_id).all()
return jsonify({'attachments': [
{
'id': a.id,
'filename': a.filename,
'stored_name': a.stored_name,
'file_size': a.file_size,
'mime_type': a.mime_type or '',
'url': url_for('admin.kb_serve_file', stored_name=a.stored_name),
'delete_url': url_for('admin.kb_delete_attachment',
article_id=article_id, att_id=a.id),
}
for a in atts
]})
# ── Create article ─────────────────────────────────────────────────────────────
@admin_bp.route('/kb/new', methods=['GET', 'POST'])
@login_required
@it_required
def kb_new():
if request.method == 'POST':
try:
article = KnowledgeBase(
title = request.form.get('title', '').strip(),
body = _sanitize_kb_body(request.form.get('body', '')),
category = request.form.get('category', ''),
tags = request.form.get('tags', ''),
author_id = current_user.id,
is_published= bool(request.form.get('is_published')) and not bool(request.form.get('_save_as_draft')),
)
db.session.add(article)
db.session.flush()
for f in request.files.getlist('attachments'):
if f and f.filename and _kb_allowed(f.filename):
att = _save_kb_file(f, article.id)
db.session.add(att)
log_action(current_user.id, 'kb_create', 'knowledge_base', article.id,
f'title={article.title}')
db.session.commit()
logger.info(f'[KB CREATE] article_id={article.id} by user_id={current_user.id}')
flash('Article created successfully.', 'success')
return jsonify({'ok': True, 'article_id': article.id, 'redirect': url_for('admin.kb_list')})
except Exception as exc:
db.session.rollback()
logger.error(f'[KB CREATE ERROR] {exc}')
return jsonify({'error': str(exc)}), 500
return render_template('admin/kb_edit.html', article=None)
# ── Edit article ───────────────────────────────────────────────────────────────
@admin_bp.route('/kb/<int:article_id>/edit', methods=['GET', 'POST'])
@login_required
@it_required
def kb_edit(article_id):
article = KnowledgeBase.query.get_or_404(article_id)
if request.method == 'POST':
try:
article.title = request.form.get('title', article.title).strip()
article.body = _sanitize_kb_body(request.form.get('body', article.body))
article.category = request.form.get('category', article.category)
article.tags = request.form.get('tags', article.tags)
article.is_published= bool(request.form.get('is_published')) and not bool(request.form.get('_save_as_draft'))
for f in request.files.getlist('attachments'):
if f and f.filename and _kb_allowed(f.filename):
att = _save_kb_file(f, article.id)
db.session.add(att)
log_action(current_user.id, 'kb_edit', 'knowledge_base', article.id)
db.session.commit()
logger.info(f'[KB EDIT] article_id={article.id} by user_id={current_user.id}')
flash('Article updated successfully.', 'success')
return jsonify({'ok': True, 'article_id': article.id, 'redirect': url_for('admin.kb_list')})
except Exception as exc:
db.session.rollback()
logger.error(f'[KB EDIT ERROR] {exc}')
return jsonify({'error': str(exc)}), 500
return render_template('admin/kb_edit.html', article=article)
@admin_bp.route('/kb/<int:article_id>/publish', methods=['POST'])
@login_required
@it_required
def kb_toggle_publish(article_id):
"""Quick publish/unpublish toggle — callable from the article list."""
article = KnowledgeBase.query.get_or_404(article_id)
article.is_published = not article.is_published
state = 'published' if article.is_published else 'unpublished'
log_action(current_user.id, f'kb_{state}', 'knowledge_base', article.id,
f'title={article.title}')
db.session.commit()
logger.info(f'[KB TOGGLE PUBLISH] article_id={article.id} is_published={article.is_published} by user_id={current_user.id}')
flash(f'Article "{article.title}" has been {state}.', 'success')
return redirect(url_for('admin.kb_list'))
@admin_bp.route('/kb/<int:article_id>/delete', methods=['POST'])
@login_required
@it_required
def kb_delete(article_id):
article = KnowledgeBase.query.get_or_404(article_id)
upload_dir = current_app.config['UPLOAD_FOLDER']
# Remove physical files before the cascade deletes the KBAttachment rows.
# Without this step the DB records disappear but the files remain on disk
# with no pointer to them — unrecoverable orphans.
for att in article.attachments.all():
filepath = os.path.join(upload_dir, att.stored_name)
if os.path.exists(filepath):
os.remove(filepath)
logger.info(f'[KB DELETE FILE] stored_name={att.stored_name} article_id={article_id} by user_id={current_user.id}')
log_action(current_user.id, 'kb_delete', 'knowledge_base', article.id,
f'title={article.title}')
logger.info(f'[KB DELETE] article_id={article.id} by user_id={current_user.id}')
db.session.delete(article)
db.session.commit()
flash('Article deleted.', 'success')
return redirect(url_for('admin.kb_list'))
# ─── Activity Log ─────────────────────────────────────────────────────────────
@admin_bp.route('/logs', methods=['GET', 'POST'])
@login_required
@admin_required
def activity_logs():
# ── Cleanup action ────────────────────────────────────────────────────────
if request.method == 'POST':
days = request.form.get('days', type=int)
if days not in (7, 30, 60, 90):
flash('Invalid retention period selected.', 'danger')
return redirect(url_for('admin.activity_logs'))
cutoff = datetime.utcnow() - timedelta(days=days)
deleted = ActivityLog.query.filter(ActivityLog.created_at < cutoff).delete()
db.session.commit()
log_action(current_user.id, 'activity_log_cleanup', 'activity_log', None,
f'deleted={deleted} older_than={days}_days cutoff={cutoff.strftime("%Y-%m-%d")}')
db.session.commit()
logger.info(f'[ACTIVITY LOG CLEANUP] deleted={deleted} days={days} by user_id={current_user.id}')
flash(f'Deleted {deleted:,} log entr{"y" if deleted == 1 else "ies"} older than {days} days.', 'success')
return redirect(url_for('admin.activity_logs'))
# ── Stats for the summary bar ─────────────────────────────────────────────
from sqlalchemy import func
total = ActivityLog.query.count()
oldest = db.session.query(func.min(ActivityLog.created_at)).scalar()
counts_by_retention = {}
for days in (7, 30, 60, 90):
cutoff = datetime.utcnow() - timedelta(days=days)
counts_by_retention[days] = ActivityLog.query.filter(
ActivityLog.created_at < cutoff
).count()
# ── Paginated log listing ─────────────────────────────────────────────────
page = request.args.get('page', 1, type=int)
logs = ActivityLog.query.order_by(ActivityLog.created_at.desc()).paginate(
page=page, per_page=50)
return render_template('admin/activity_logs.html',
logs = logs,
total = total,
oldest = oldest,
counts_by_retention= counts_by_retention,
)
def _roles():
return [UserRole.EMPLOYEE, UserRole.IT_STAFF, UserRole.ADMIN]