466 lines
21 KiB
Python
466 lines
21 KiB
Python
import logging
|
|
import os
|
|
import uuid
|
|
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():
|
|
articles = KnowledgeBase.query.order_by(KnowledgeBase.created_at.desc()).all()
|
|
return render_template('admin/kb_list.html', articles=articles)
|
|
|
|
|
|
# ── 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')
|
|
@login_required
|
|
@admin_required
|
|
def activity_logs():
|
|
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)
|
|
|
|
|
|
def _roles():
|
|
return [UserRole.EMPLOYEE, UserRole.IT_STAFF, UserRole.ADMIN] |