4529 lines
193 KiB
Python
4529 lines
193 KiB
Python
import os
|
||
# reload
|
||
import sys
|
||
import shutil
|
||
import time
|
||
from typing import List, Optional
|
||
from fastapi import FastAPI, File, UploadFile, Form, Request, HTTPException, Depends, status, Cookie, BackgroundTasks
|
||
from fastapi.responses import HTMLResponse, FileResponse, RedirectResponse, Response
|
||
from fastapi.staticfiles import StaticFiles
|
||
from fastapi.templating import Jinja2Templates
|
||
from fastapi.security import HTTPBasic, HTTPBasicCredentials
|
||
from werkzeug.utils import secure_filename
|
||
import audit_decryption
|
||
from colorama import Fore
|
||
import uvicorn
|
||
import secrets
|
||
from datetime import datetime, timedelta
|
||
import json
|
||
import socket
|
||
import struct
|
||
import hashlib
|
||
import random
|
||
import uuid
|
||
import base64
|
||
import ipaddress
|
||
import re
|
||
import requests as http_requests
|
||
from authenticator.radius_login import check_radius_login, check_radius_login_extended
|
||
|
||
# Database imports
|
||
import models
|
||
from models import UserEmailRecord, UserFileRecord
|
||
from database import engine, SessionLocal
|
||
|
||
# Email notification
|
||
# EMAIL_MODULE_AVAILABLE: True nếu import thành công, False nếu thiếu thư viện
|
||
# email_enabled_runtime: trạng thái bật/tắt có thể thay đổi tại runtime bởi admin
|
||
try:
|
||
from email_notifier import send_processing_complete_email
|
||
EMAIL_MODULE_AVAILABLE = True
|
||
email_enabled_runtime = False # Mặc định TẮT; admin bật qua trang quản lý
|
||
print(Fore.GREEN + "[EMAIL] Email notification module loaded (disabled by default)" + Fore.RESET)
|
||
except ImportError:
|
||
EMAIL_MODULE_AVAILABLE = False
|
||
email_enabled_runtime = False
|
||
print(Fore.YELLOW + "[EMAIL] Email notification module not available" + Fore.RESET)
|
||
|
||
# Alias để tương thích với code cũ (sẽ đọc runtime state qua helper)
|
||
def _email_enabled() -> bool:
|
||
"""Kiểm tra email có thực sự khả dụng và được bật không."""
|
||
return EMAIL_MODULE_AVAILABLE and email_enabled_runtime
|
||
|
||
app = FastAPI(title="Audit Hardening Tool", redirect_slashes=False)
|
||
security = HTTPBasic()
|
||
|
||
# Add middleware to handle X-Forwarded-* headers from reverse proxy (Caddy)
|
||
from uvicorn.middleware.proxy_headers import ProxyHeadersMiddleware
|
||
app.add_middleware(ProxyHeadersMiddleware, trusted_hosts="*")
|
||
|
||
# Database initialization
|
||
models.Base.metadata.create_all(bind=engine)
|
||
|
||
@app.on_event("startup")
|
||
async def startup_event():
|
||
print(Fore.CYAN + "[STARTUP] Initializing and syncing database..." + Fore.RESET)
|
||
try:
|
||
extract_info_from_output_files()
|
||
print(Fore.GREEN + "[STARTUP] Database synchronization complete." + Fore.RESET)
|
||
except Exception as e:
|
||
print(Fore.RED + f"[STARTUP] Error syncing database: {e}" + Fore.RESET)
|
||
|
||
# Mount static files for icon
|
||
app.mount("/icon", StaticFiles(directory=os.path.join(os.path.dirname(__file__), "icon")), name="icon")
|
||
|
||
# Load users from config file
|
||
def load_users():
|
||
"""Load users from users_config.json"""
|
||
try:
|
||
with open('users_config.json', 'r', encoding='utf-8') as f:
|
||
config = json.load(f)
|
||
return config.get('users', {}), config.get('settings', {})
|
||
except FileNotFoundError:
|
||
print(Fore.YELLOW + "[WARNING] users_config.json not found. Using default users." + Fore.RESET)
|
||
return {
|
||
"admin": {
|
||
"username": "admin",
|
||
"password": "admin123",
|
||
"role": "admin",
|
||
"full_name": "Administrator",
|
||
"can_view_outputs": True,
|
||
"can_download": True
|
||
}
|
||
}, {"require_auth_for_outputs": True, "require_auth_for_download": True}
|
||
except Exception as e:
|
||
print(Fore.RED + f"[ERROR] Failed to load users_config.json: {e}" + Fore.RESET)
|
||
return {}, {}
|
||
|
||
AUTHORIZED_USERS, AUTH_SETTINGS = load_users()
|
||
|
||
# Load RADIUS configuration
|
||
def load_radius_config():
|
||
"""Load RADIUS configuration from radius_config.json"""
|
||
try:
|
||
with open('radius_config.json', 'r', encoding='utf-8') as f:
|
||
config = json.load(f)
|
||
print(Fore.GREEN + f"[CONFIG] RADIUS configuration loaded. Enabled: {config.get('enabled', False)}" + Fore.RESET)
|
||
return config
|
||
except FileNotFoundError:
|
||
print(Fore.YELLOW + "[WARNING] radius_config.json not found. RADIUS authentication disabled." + Fore.RESET)
|
||
return {'enabled': False}
|
||
except Exception as e:
|
||
print(Fore.RED + f"[ERROR] Failed to load radius_config.json: {e}" + Fore.RESET)
|
||
return {'enabled': False}
|
||
|
||
RADIUS_CONFIG = load_radius_config()
|
||
|
||
# Load SSO configuration
|
||
def load_sso_config():
|
||
"""Load SSO Portal configuration from sso_config.json"""
|
||
try:
|
||
sso_path = 'sso_config.json' if os.path.exists('sso_config.json') else os.path.join(os.path.dirname(os.path.dirname(__file__)), 'sso_config.json')
|
||
with open(sso_path, 'r', encoding='utf-8') as f:
|
||
config = json.load(f)
|
||
return config
|
||
except Exception as e:
|
||
return {'enabled': False}
|
||
|
||
SSO_CONFIG = load_sso_config()
|
||
|
||
# Session storage (in production, use Redis or database)
|
||
# Format: {session_id: {'files': [list of filenames], 'created_at': datetime, 'username': str or None}}
|
||
sessions = {}
|
||
processing_tasks = {}
|
||
|
||
# Session timeout (hours)
|
||
SESSION_TIMEOUT_HOURS = 24
|
||
|
||
# Configuration
|
||
UPLOAD_FOLDER = 'uploads'
|
||
OUTPUT_FOLDER = 'output'
|
||
CONFIG_FOLDER = 'config'
|
||
KEYS_FOLDER = 'keys'
|
||
ALLOWED_EXTENSIONS = {'enc', 'txt'}
|
||
|
||
# OS Configuration mapping
|
||
OS_CONFIG = {
|
||
'windows': {
|
||
'checklist': 'Windows_Checklist.xlsx',
|
||
'config': 'windows_config.json',
|
||
'name': 'Windows'
|
||
},
|
||
'centos': {
|
||
'checklist': 'CentOS_Checklist_NEW.xlsx',
|
||
'config': 'centos_config_new.json',
|
||
'name': 'CentOS'
|
||
},
|
||
'rhel': {
|
||
'checklist': 'RHEL_Checklist_NEW.xlsx',
|
||
'config': 'rhel_config_new.json',
|
||
'name': 'RHEL (Red Hat Enterprise Linux)'
|
||
},
|
||
'ubuntu': {
|
||
'checklist': 'Ubuntu_Checklist_NEW.xlsx',
|
||
'config': 'ubuntu_config_new.json',
|
||
'name': 'Ubuntu'
|
||
},
|
||
'oracle': {
|
||
'checklist': 'Oracle_Linux_Checklist.xlsx',
|
||
'config': 'oracle_linux_config.json',
|
||
'name': 'Oracle Linux'
|
||
}
|
||
}
|
||
|
||
def detect_os_from_filename(filename: str) -> Optional[str]:
|
||
"""
|
||
Detect OS type from filename pattern.
|
||
Filename patterns like: localhost.localdomain_centos_linux_7_IP-Address_-10.144.8.226.txt.enc
|
||
Returns the OS key (centos, ubuntu, rhel, oracle, windows) or None if not detected.
|
||
"""
|
||
if not filename:
|
||
return None
|
||
|
||
filename_lower = filename.lower()
|
||
|
||
# Define OS detection patterns - order matters (more specific first)
|
||
# NOTE: 'ol_' đã bị loại bỏ vì gây false positive (ví dụ 'protocol_data' match oracle)
|
||
os_patterns = {
|
||
'centos': ['centos', 'centos_linux', 'centos linux'],
|
||
'rhel': ['rhel', 'red_hat', 'red hat', 'redhat', 'red-hat'],
|
||
'ubuntu': ['ubuntu'],
|
||
'oracle': ['oracle', 'oracle_linux', 'oracle linux'],
|
||
'windows': ['windows', 'win_server', 'win-server', 'winserver']
|
||
}
|
||
|
||
for os_key, patterns in os_patterns.items():
|
||
for pattern in patterns:
|
||
if pattern in filename_lower:
|
||
print(Fore.CYAN + f"[AUTO-DETECT] Detected OS '{os_key}' from filename: {filename}" + Fore.RESET)
|
||
return os_key
|
||
|
||
print(Fore.YELLOW + f"[AUTO-DETECT] Could not detect OS from filename: {filename}" + Fore.RESET)
|
||
return None
|
||
|
||
def detect_os_from_multiple_files(filenames: List[str]) -> tuple[Optional[str], bool]:
|
||
"""
|
||
Detect OS from multiple file names.
|
||
Returns: (detected_os, all_same) where all_same indicates if all files have the same OS.
|
||
"""
|
||
detected_os_list = []
|
||
|
||
for filename in filenames:
|
||
os_type = detect_os_from_filename(filename)
|
||
if os_type:
|
||
detected_os_list.append(os_type)
|
||
|
||
if not detected_os_list:
|
||
return None, True
|
||
|
||
# Check if all detected OS are the same
|
||
unique_os = set(detected_os_list)
|
||
if len(unique_os) == 1:
|
||
return detected_os_list[0], True
|
||
else:
|
||
# Mixed OS detected — log cảnh báo
|
||
print(Fore.YELLOW + f"[AUTO-DETECT] ⚠ Mixed OS detected in batch: {unique_os}" + Fore.RESET)
|
||
return detected_os_list[0], False
|
||
|
||
def detect_os_from_decrypted_file(enc_path: str) -> Optional[str]:
|
||
"""
|
||
Decrypt only the first few blocks (~5KB) of the .enc file to detect OS from content.
|
||
Extremely fast (< 0.1s) compared to full file decryption (300s+).
|
||
"""
|
||
try:
|
||
dec_filename = os.path.basename(enc_path).replace('.enc', '')
|
||
|
||
# Nếu file giải mã đã tồn tại sẵn trên đĩa -> đọc trực tiếp
|
||
if os.path.exists(dec_filename):
|
||
with open(dec_filename, 'r', encoding='utf-8', errors='replace') as fh:
|
||
lines = fh.read().splitlines()
|
||
return audit_decryption.detect_os_from_content(lines)
|
||
|
||
# Giải mã siêu nhanh N block đầu tiên trong RAM
|
||
header_lines = audit_decryption.decryption_read_header(enc_path, max_blocks=30)
|
||
return audit_decryption.detect_os_from_content(header_lines)
|
||
|
||
except Exception as e:
|
||
print(Fore.YELLOW + f"[CONTENT-DETECT] Could not detect OS from content: {e}" + Fore.RESET)
|
||
|
||
return None
|
||
|
||
|
||
# Create necessary folders
|
||
os.makedirs(UPLOAD_FOLDER, exist_ok=True)
|
||
os.makedirs(OUTPUT_FOLDER, exist_ok=True)
|
||
|
||
# Setup templates with Starlette 1.0 compatibility
|
||
# Starlette 1.0 changed TemplateResponse(name, context) to TemplateResponse(request, name, context)
|
||
class CompatTemplates(Jinja2Templates):
|
||
"""Wrapper to support old-style TemplateResponse(name, {request: ..., ...}) on Starlette 1.0+"""
|
||
def TemplateResponse(self, name, context=None, **kwargs):
|
||
if context is None:
|
||
context = {}
|
||
# Extract request from context (old-style API)
|
||
request = context.pop("request", None)
|
||
if request is None:
|
||
request = kwargs.pop("request", None)
|
||
if request is None:
|
||
raise ValueError("request is required in TemplateResponse context")
|
||
if "docs_list" not in context and "get_available_docs" in globals():
|
||
context["docs_list"] = get_available_docs()
|
||
# Call new-style API: super().TemplateResponse(request, name, context, ...)
|
||
return super().TemplateResponse(request=request, name=name, context=context, **kwargs)
|
||
|
||
templates = CompatTemplates(directory=os.path.join(os.path.dirname(__file__), "templates"))
|
||
|
||
def allowed_file(filename: str) -> bool:
|
||
return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
|
||
|
||
# ────────────────────────────────────────────────────────────────────────────
|
||
# Email registration helpers
|
||
# ────────────────────────────────────────────────────────────────────────────
|
||
_VNPT_EMAIL_RE = re.compile(r'^[a-zA-Z0-9._%+\-]+@vnpt\.vn$', re.IGNORECASE)
|
||
|
||
def validate_vnpt_email(email: str) -> bool:
|
||
"""Kiểm tra email có đúng định dạng @vnpt.vn."""
|
||
return bool(_VNPT_EMAIL_RE.match(email.strip()))
|
||
|
||
def record_email_upload(email: str, num_files: int):
|
||
"""
|
||
Cập nhật (hoặc tạo mới) bản ghi UserEmailRecord cho email.
|
||
Tăng total_uploads += 1, total_files += num_files, cập nhật last_seen.
|
||
"""
|
||
if not email:
|
||
return
|
||
db = SessionLocal()
|
||
try:
|
||
record = db.query(UserEmailRecord).filter(UserEmailRecord.email == email.lower()).first()
|
||
now = datetime.utcnow()
|
||
if record:
|
||
record.total_uploads += 1
|
||
record.total_files += num_files
|
||
record.last_seen = now
|
||
else:
|
||
display = email.split('@')[0]
|
||
record = UserEmailRecord(
|
||
email=email.lower(),
|
||
display_name=display,
|
||
total_uploads=1,
|
||
total_files=num_files,
|
||
first_seen=now,
|
||
last_seen=now,
|
||
)
|
||
db.add(record)
|
||
db.commit()
|
||
print(Fore.CYAN + f"[EMAIL-STAT] Recorded upload for {email}: +{num_files} file(s)" + Fore.RESET)
|
||
except Exception as e:
|
||
db.rollback()
|
||
print(Fore.RED + f"[EMAIL-STAT] DB error: {e}" + Fore.RESET)
|
||
finally:
|
||
db.close()
|
||
|
||
def get_or_create_session(request: Request, response: Response = None) -> str:
|
||
"""
|
||
Lấy hoặc tạo session ID từ cookie.
|
||
Nếu session ID còn trong cookie nhưng dữ liệu RAM đã mất (do server restart),
|
||
tự động tạo lại session và khôi phục user_info từ AUTHORIZED_USERS nếu có thể.
|
||
"""
|
||
session_id = request.cookies.get('session_id')
|
||
|
||
if not session_id or session_id not in sessions:
|
||
# Tạo session mới
|
||
session_id = str(uuid.uuid4())
|
||
sessions[session_id] = {
|
||
'files': [],
|
||
'created_at': datetime.now(),
|
||
'username': None,
|
||
'user_info': None,
|
||
}
|
||
print(Fore.CYAN + f"[SESSION] Created new session: {session_id}" + Fore.RESET)
|
||
|
||
# Cố gắng phục hồi user_info từ cookie 'logged_in_user' (nếu server đã restart)
|
||
saved_username = request.cookies.get('logged_in_user')
|
||
if saved_username:
|
||
if saved_username in AUTHORIZED_USERS:
|
||
# User có trong whitelist → phục hồi đầy đủ từ config
|
||
restored = dict(AUTHORIZED_USERS[saved_username])
|
||
|
||
# Apply unit mapping if missing
|
||
if not restored.get('unit_id') or not restored.get('unit_name'):
|
||
override_name, override_id = get_fixed_unit_override(saved_username)
|
||
if override_id:
|
||
restored['unit_id'] = override_id
|
||
restored['unit_name'] = override_name
|
||
else:
|
||
m_name, m_id = get_unit_info_from_username(saved_username)
|
||
if m_id:
|
||
restored['unit_id'] = m_id
|
||
restored['unit_name'] = m_name
|
||
|
||
sessions[session_id]['username'] = saved_username
|
||
sessions[session_id]['user_info'] = restored
|
||
sessions[session_id]['unit_id'] = str(restored.get('unit_id', ''))
|
||
sessions[session_id]['unit_name'] = restored.get('unit_name', '')
|
||
sessions[session_id]['fullname'] = restored.get('full_name', saved_username)
|
||
print(Fore.YELLOW + f"[SESSION] Restored user_info for whitelist user '{saved_username}' from config" + Fore.RESET)
|
||
else:
|
||
# RADIUS user không có trong whitelist → phục hồi từ các cookie bổ sung
|
||
saved_unit_id = request.cookies.get('user_unit_id', '')
|
||
saved_unit_name = request.cookies.get('user_unit_name', '')
|
||
saved_role = request.cookies.get('user_role', 'user')
|
||
restored = {
|
||
'username': saved_username,
|
||
'full_name': saved_username,
|
||
'role': saved_role,
|
||
'unit_id': saved_unit_id,
|
||
'unit_name': saved_unit_name,
|
||
'auth_method': 'radius',
|
||
'can_view_outputs': True,
|
||
'can_download': True,
|
||
}
|
||
sessions[session_id]['username'] = saved_username
|
||
sessions[session_id]['user_info'] = restored
|
||
sessions[session_id]['unit_id'] = saved_unit_id
|
||
sessions[session_id]['unit_name'] = saved_unit_name
|
||
sessions[session_id]['fullname'] = saved_username
|
||
print(Fore.YELLOW + f"[SESSION] Restored partial user_info for RADIUS user '{saved_username}' from cookies" + Fore.RESET)
|
||
|
||
# Cleanup old sessions
|
||
cleanup_old_sessions()
|
||
|
||
return session_id
|
||
|
||
def cleanup_old_sessions():
|
||
"""
|
||
Xóa các session cũ hơn SESSION_TIMEOUT_HOURS
|
||
"""
|
||
now = datetime.now()
|
||
expired_sessions = []
|
||
|
||
for session_id, session_data in sessions.items():
|
||
if (now - session_data['created_at']).total_seconds() > SESSION_TIMEOUT_HOURS * 3600:
|
||
expired_sessions.append(session_id)
|
||
|
||
for session_id in expired_sessions:
|
||
print(Fore.YELLOW + f"[SESSION] Expired session removed: {session_id}" + Fore.RESET)
|
||
del sessions[session_id]
|
||
|
||
def add_file_to_session(session_id: str, filename: str):
|
||
"""
|
||
Thêm file vào session của user.
|
||
Đồng thời lưu vào cơ sở dữ liệu nếu user đã đăng nhập để quản lý theo người dùng.
|
||
"""
|
||
if session_id in sessions:
|
||
if filename not in sessions[session_id]['files']:
|
||
sessions[session_id]['files'].append(filename)
|
||
print(Fore.GREEN + f"[SESSION] Added file '{filename}' to session {session_id}" + Fore.RESET)
|
||
|
||
# Lưu vào Database UserFileRecord để quản lý theo user
|
||
user_info = sessions[session_id].get('user_info')
|
||
if user_info:
|
||
try:
|
||
db = SessionLocal()
|
||
existing = db.query(UserFileRecord).filter_by(
|
||
filename=filename, username=user_info['username']
|
||
).first()
|
||
if not existing:
|
||
new_record = UserFileRecord(
|
||
filename=filename,
|
||
username=user_info['username'],
|
||
unit_id=user_info.get('unit_id', ''),
|
||
unit_name=user_info.get('unit_name', '')
|
||
)
|
||
db.add(new_record)
|
||
db.commit()
|
||
print(Fore.GREEN + f"[DB] File '{filename}' linked to user '{user_info['username']}'" + Fore.RESET)
|
||
except Exception as e:
|
||
print(Fore.RED + f"[DB ERROR] Could not link file to user: {e}" + Fore.RESET)
|
||
finally:
|
||
db.close()
|
||
|
||
def get_app_root_path(request: Request) -> str:
|
||
"""
|
||
Lấy root_path chính xác của ứng dụng khi chạy sau Reverse Proxy.
|
||
"""
|
||
root = request.scope.get("root_path", "").rstrip("/")
|
||
if root:
|
||
return root
|
||
fwd_prefix = request.headers.get("x-forwarded-prefix", "").rstrip("/")
|
||
if fwd_prefix:
|
||
return fwd_prefix
|
||
import sys
|
||
for i, arg in enumerate(sys.argv):
|
||
if arg == "--root-path" and i + 1 < len(sys.argv):
|
||
return sys.argv[i+1].rstrip("/")
|
||
if arg.startswith("--root-path="):
|
||
return arg.split("=", 1)[1].rstrip("/")
|
||
referer = request.headers.get("referer", "")
|
||
if "/audit/" in referer or "/audit" in referer:
|
||
return "/audit"
|
||
return ""
|
||
|
||
def can_access_file(session_id: str, filename: str) -> bool:
|
||
"""
|
||
Kiểm tra xem session hoặc user hiện tại có quyền truy cập file không.
|
||
- Admin có quyền truy cập TẤT CẢ các file.
|
||
- User thường xem được các file do mình sở hữu (trong UserFileRecord) hoặc trong session hiện tại.
|
||
"""
|
||
if session_id in sessions:
|
||
user_info = sessions[session_id].get('user_info')
|
||
if user_info:
|
||
# Admin có toàn quyền
|
||
if user_info.get('role') == 'admin':
|
||
return True
|
||
|
||
# Kiểm tra theo user trong database
|
||
try:
|
||
db = SessionLocal()
|
||
record = db.query(UserFileRecord).filter_by(
|
||
filename=filename, username=user_info['username']
|
||
).first()
|
||
if record:
|
||
return True
|
||
except Exception:
|
||
pass
|
||
finally:
|
||
db.close()
|
||
|
||
# Fallback kiểm tra session (cho guest)
|
||
return filename in sessions[session_id]['files']
|
||
return False
|
||
|
||
# ────────────────────────────────────────────────────────────────────────────
|
||
# privacyIDEA helper — tra cứu đơn vị user sau xác thực RADIUS
|
||
# ────────────────────────────────────────────────────────────────────────────
|
||
_pi_admin_token_cache: dict = {} # {token, expires_at}
|
||
|
||
def _pi_get_admin_token() -> str | None:
|
||
"""Lấy hoặc tái sử dụng admin JWT token từ privacyIDEA."""
|
||
import time as _time
|
||
import urllib.request as _ureq
|
||
import urllib.parse as _uparse
|
||
import ssl as _ssl
|
||
import json as _json
|
||
|
||
pi_cfg = RADIUS_CONFIG.get('privacyidea', {})
|
||
if not pi_cfg.get('enabled', False):
|
||
return None
|
||
|
||
# Kiểm tra cache còn hạn không (token tồn tại ~1h, cache 50 phút)
|
||
cached = _pi_admin_token_cache
|
||
if cached.get('token') and cached.get('expires_at', 0) > _time.time():
|
||
return cached['token']
|
||
|
||
url = pi_cfg['url'].rstrip('/') + '/auth'
|
||
admin = pi_cfg.get('admin_user', 'admin')
|
||
password = pi_cfg.get('admin_password', '')
|
||
timeout = pi_cfg.get('timeout', 5)
|
||
verify = pi_cfg.get('verify_ssl', False)
|
||
|
||
if not password:
|
||
print(Fore.YELLOW + "[PI] privacyIDEA admin_password chưa được cấu hình trong radius_config.json" + Fore.RESET)
|
||
return None
|
||
|
||
body = _uparse.urlencode({'username': admin, 'password': password}).encode()
|
||
ctx = _ssl.create_default_context()
|
||
if not verify:
|
||
ctx.check_hostname = False
|
||
ctx.verify_mode = _ssl.CERT_NONE
|
||
|
||
try:
|
||
req = _ureq.Request(url, data=body, method='POST')
|
||
req.add_header('Content-Type', 'application/x-www-form-urlencoded')
|
||
with _ureq.urlopen(req, context=ctx, timeout=timeout) as resp:
|
||
result = _json.loads(resp.read().decode())
|
||
token = result.get('result', {}).get('value', {}).get('token')
|
||
if token:
|
||
_pi_admin_token_cache['token'] = token
|
||
_pi_admin_token_cache['expires_at'] = _time.time() + 3000 # ~50 phút
|
||
print(Fore.CYAN + "[PI] Admin token refreshed" + Fore.RESET)
|
||
return token
|
||
except Exception as e:
|
||
print(Fore.YELLOW + f"[PI] Không lấy được admin token: {e}" + Fore.RESET)
|
||
return None
|
||
|
||
|
||
def get_groups_from_privacyidea(username: str) -> tuple[list, str]:
|
||
"""
|
||
Tra cứu groups của user từ privacyIDEA sau khi RADIUS xác thực thành công.
|
||
Returns: (groups: list, full_name: str)
|
||
Nếu lỗi hoặc không tìm thấy, trả về ([], '')
|
||
"""
|
||
import urllib.request as _ureq
|
||
import urllib.parse as _uparse
|
||
import ssl as _ssl
|
||
import json as _json
|
||
|
||
pi_cfg = RADIUS_CONFIG.get('privacyidea', {})
|
||
if not pi_cfg.get('enabled', False):
|
||
return [], ''
|
||
|
||
token = _pi_get_admin_token()
|
||
if not token:
|
||
return [], ''
|
||
|
||
realm = pi_cfg.get('realm', '')
|
||
base = pi_cfg['url'].rstrip('/')
|
||
timeout = pi_cfg.get('timeout', 5)
|
||
verify = pi_cfg.get('verify_ssl', False)
|
||
|
||
params = {'username': username}
|
||
if realm:
|
||
params['realm'] = realm
|
||
url = base + '/user/?' + _uparse.urlencode(params)
|
||
|
||
ctx = _ssl.create_default_context()
|
||
if not verify:
|
||
ctx.check_hostname = False
|
||
ctx.verify_mode = _ssl.CERT_NONE
|
||
|
||
try:
|
||
req = _ureq.Request(url, method='GET')
|
||
req.add_header('Authorization', token)
|
||
req.add_header('PI-Authorization', token)
|
||
with _ureq.urlopen(req, context=ctx, timeout=timeout) as resp:
|
||
data = _json.loads(resp.read().decode())
|
||
users = data.get('result', {}).get('value', [])
|
||
if not users:
|
||
print(Fore.YELLOW + f"[PI] Không tìm thấy user '{username}' trong privacyIDEA" + Fore.RESET)
|
||
return [], ''
|
||
|
||
user = users[0]
|
||
groups = user.get('groups', [])
|
||
givenname = user.get('givenname', '')
|
||
surname = user.get('surname', '').replace('- ', '').strip()
|
||
full_name = f"{givenname} {surname}".strip() or username
|
||
|
||
print(Fore.CYAN + f"[PI] User '{username}' → full_name='{full_name}', groups={len(groups)}" + Fore.RESET)
|
||
return groups, full_name
|
||
|
||
except Exception as e:
|
||
print(Fore.YELLOW + f"[PI] Lỗi tra cứu user '{username}': {e}" + Fore.RESET)
|
||
return [], ''
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# Unit Mapping — tra cứu đơn vị từ file unit_mapping.json (không cần server)
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
_UNIT_MAPPING_PATH = 'unit_mapping.json' if os.path.exists('unit_mapping.json') else os.path.join(os.path.dirname(os.path.dirname(__file__)), 'unit_mapping.json')
|
||
_unit_mapping_cache: dict = {}
|
||
_unit_mapping_mtime: float = 0.0
|
||
|
||
def _load_unit_mapping() -> dict:
|
||
global _unit_mapping_cache, _unit_mapping_mtime
|
||
try:
|
||
mtime = os.path.getmtime(_UNIT_MAPPING_PATH)
|
||
if mtime != _unit_mapping_mtime:
|
||
with open(_UNIT_MAPPING_PATH, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
_unit_mapping_cache = data
|
||
_unit_mapping_mtime = mtime
|
||
print(Fore.CYAN + "[UNIT-MAP] Loaded unit_mapping.json" + Fore.RESET)
|
||
except Exception as e:
|
||
if not isinstance(e, FileNotFoundError):
|
||
print(Fore.YELLOW + f"[UNIT-MAP] Error reading unit_mapping.json: {e}" + Fore.RESET)
|
||
return _unit_mapping_cache
|
||
|
||
def get_unit_info_from_username(username: str) -> tuple[str, str]:
|
||
"""Trả về (unit_name, unit_id) dựa vào username."""
|
||
mapping = _load_unit_mapping()
|
||
users = mapping.get('users', {})
|
||
units = mapping.get('units', {})
|
||
|
||
unit_id_val = users.get(username)
|
||
if unit_id_val is None or unit_id_val == "":
|
||
return '', ''
|
||
|
||
unit_id = str(unit_id_val)
|
||
return units.get(unit_id, ''), unit_id
|
||
|
||
def get_fixed_unit_override(username: str) -> tuple[str, str]:
|
||
"""
|
||
Kiểm tra xem user có trong danh sách 'fixed_unit_override' không.
|
||
Nếu có → trả về (unit_name, unit_id), GHI ĐÈ hoàn toàn đơn vị từ LDAP/SSO.
|
||
Nếu không → trả về ('', '').
|
||
"""
|
||
mapping = _load_unit_mapping()
|
||
overrides = mapping.get('fixed_unit_override', {})
|
||
units = mapping.get('units', {})
|
||
|
||
unit_id_val = overrides.get(username)
|
||
if unit_id_val is None or unit_id_val == "":
|
||
return '', ''
|
||
|
||
unit_id = str(unit_id_val)
|
||
unit_name = units.get(unit_id, '')
|
||
if unit_name:
|
||
print(Fore.MAGENTA + f"[UNIT-OVERRIDE] User '{username}' → đơn vị cố định: '{unit_name}' (ID: {unit_id})" + Fore.RESET)
|
||
return unit_name, unit_id
|
||
|
||
def map_groups_to_unit(groups: list) -> tuple[str, str]:
|
||
"""
|
||
Duyệt danh sách các group của user, nếu group có chứa từ khóa
|
||
trong group_mapping thì trả về (unit_name, unit_id).
|
||
"""
|
||
mapping = _load_unit_mapping()
|
||
group_map = mapping.get('group_mapping', {})
|
||
units = mapping.get('units', {})
|
||
|
||
for dn in groups:
|
||
if not isinstance(dn, str): continue
|
||
dn_lower = dn.lower()
|
||
for kw, uid in group_map.items():
|
||
if kw.lower() in dn_lower:
|
||
unit_id = str(uid)
|
||
return units.get(unit_id, ''), unit_id
|
||
return '', ''
|
||
|
||
def is_admin_in_mapping(username: str) -> bool:
|
||
"""Kiểm tra xem user có nằm trong danh sách admins của unit_mapping.json hay không."""
|
||
mapping = _load_unit_mapping()
|
||
admins = mapping.get('admins', [])
|
||
return username in admins
|
||
|
||
|
||
def authenticate_radius(username: str, password: str) -> tuple[bool, str, dict]:
|
||
"""
|
||
Xác thực người dùng qua RADIUS server sử dụng module authenticator.radius_login
|
||
Returns: (success: bool, message: str, radius_attrs: dict)
|
||
radius_attrs có thể chứa: filter_id, reply_message, class, raw
|
||
"""
|
||
if not RADIUS_CONFIG.get('enabled', False):
|
||
return False, "RADIUS authentication is disabled", {}
|
||
|
||
try:
|
||
server = RADIUS_CONFIG['server']
|
||
port = RADIUS_CONFIG.get('port', 1812)
|
||
secret = RADIUS_CONFIG['secret']
|
||
realm = RADIUS_CONFIG.get('realm', '')
|
||
timeout = RADIUS_CONFIG.get('timeout', 5)
|
||
|
||
radius_username = f"{username}{realm}" if realm else username
|
||
|
||
print(Fore.CYAN + f"[RADIUS] Attempting authentication for user: {radius_username}" + Fore.RESET)
|
||
print(Fore.CYAN + f"[RADIUS] Server: {server}:{port}" + Fore.RESET)
|
||
|
||
start_time = time.time()
|
||
ok, attrs = check_radius_login_extended(server, port, secret, radius_username, password, timeout)
|
||
elapsed = time.time() - start_time
|
||
|
||
if ok:
|
||
print(Fore.GREEN + f"[RADIUS] ✓ Authentication successful for user: {username} ({elapsed:.1f}s)" + Fore.RESET)
|
||
if attrs.get('filter_id'):
|
||
print(Fore.CYAN + f"[RADIUS] Filter-Id: {attrs['filter_id']}" + Fore.RESET)
|
||
return True, "RADIUS authentication successful", attrs
|
||
else:
|
||
if elapsed >= (timeout - 0.5):
|
||
print(Fore.RED + f"[RADIUS] ✗ Timeout ({elapsed:.1f}s) for user: {username}" + Fore.RESET)
|
||
return False, "RADIUS timeout", {}
|
||
else:
|
||
print(Fore.YELLOW + f"[RADIUS] ✗ Authentication rejected for user: {username} ({elapsed:.1f}s)" + Fore.RESET)
|
||
return False, "RADIUS authentication rejected", {}
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
print(Fore.RED + f"[RADIUS] ✗ Error: {str(e)}" + Fore.RESET)
|
||
print(Fore.RED + f"[RADIUS] Traceback: {traceback.format_exc()}" + Fore.RESET)
|
||
return False, f"RADIUS error: {str(e)}", {}
|
||
|
||
def authenticate_local(username: str, password: str) -> tuple[bool, Optional[dict]]:
|
||
"""
|
||
Xác thực người dùng qua local users_config.json
|
||
Returns: (success: bool, user_info: dict or None)
|
||
"""
|
||
if username not in AUTHORIZED_USERS:
|
||
return False, None
|
||
|
||
user = AUTHORIZED_USERS[username]
|
||
if secrets.compare_digest(password.encode('utf-8'), user['password'].encode('utf-8')):
|
||
print(Fore.GREEN + f"[LOCAL] Authentication successful for user: {username}" + Fore.RESET)
|
||
return True, user
|
||
|
||
return False, None
|
||
|
||
def verify_credentials(credentials: HTTPBasicCredentials = Depends(security)):
|
||
"""
|
||
Verify user credentials - Hybrid RADIUS + Local fallback
|
||
1. Thử xác thực qua RADIUS server trước
|
||
2. Nếu RADIUS thất bại hoặc không khả dụng, fallback sang local users_config.json
|
||
"""
|
||
username = credentials.username
|
||
password = credentials.password
|
||
|
||
print(Fore.BLUE + f"[AUTH] Login attempt for user: {username}" + Fore.RESET)
|
||
|
||
# Bước 1: Thử RADIUS authentication trước
|
||
if RADIUS_CONFIG['enabled']:
|
||
radius_success, radius_message = authenticate_radius(username, password)
|
||
|
||
if radius_success:
|
||
# RADIUS authentication thành công - kiểm tra whitelist
|
||
if username in AUTHORIZED_USERS:
|
||
user = AUTHORIZED_USERS[username]
|
||
print(Fore.GREEN + f"[AUTH] RADIUS user logged in: {username} ({user.get('role', 'user')})" + Fore.RESET)
|
||
return user
|
||
else:
|
||
print(Fore.RED + f"[AUTH] RADIUS OK but user '{username}' NOT in whitelist" + Fore.RESET)
|
||
raise HTTPException(
|
||
status_code=status.HTTP_403_FORBIDDEN,
|
||
detail=f"Tài khoản '{username}' chưa được cấp quyền truy cập",
|
||
)
|
||
else:
|
||
print(Fore.YELLOW + f"[AUTH] RADIUS authentication failed: {radius_message}" + Fore.RESET)
|
||
|
||
# Bước 2: Fallback sang Local authentication
|
||
print(Fore.CYAN + f"[AUTH] Trying local authentication for user: {username}" + Fore.RESET)
|
||
local_success, user = authenticate_local(username, password)
|
||
|
||
if local_success:
|
||
print(Fore.GREEN + f"[AUTH] Local user logged in: {username} ({user.get('role', 'unknown')})" + Fore.RESET)
|
||
return user
|
||
|
||
# Cả RADIUS và Local đều thất bại
|
||
print(Fore.RED + f"[AUTH] Authentication failed for user: {username}" + Fore.RESET)
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="Invalid username or password",
|
||
headers={"WWW-Authenticate": "Basic"},
|
||
)
|
||
|
||
def check_permission(user: dict, permission: str) -> bool:
|
||
"""Check if user has specific permission"""
|
||
return user.get(permission, False)
|
||
|
||
def get_session_user(request: Request) -> Optional[dict]:
|
||
"""Lấy user từ session cookie (đã đăng nhập qua form)"""
|
||
session_id = request.cookies.get('session_id')
|
||
if session_id and session_id in sessions:
|
||
user_info = sessions[session_id].get('user_info')
|
||
if user_info:
|
||
return user_info
|
||
return None
|
||
|
||
async def verify_credentials_or_session(request: Request) -> dict:
|
||
"""
|
||
Xác thực user theo thứ tự:
|
||
1. Session cookie (đã login qua form / SSO) - không popup
|
||
2. Nếu không có session → redirect về trang login
|
||
Login là BẮT BUỘC với mọi route được bảo vệ.
|
||
"""
|
||
user = get_session_user(request)
|
||
if user:
|
||
return user
|
||
|
||
# Không có session → redirect về login
|
||
# Lưu lại URL hiện tại vào query param next= để sau login có thể quay lại
|
||
next_url = str(request.url)
|
||
login_url = str(request.url_for('login')) + f"?next={next_url}"
|
||
raise HTTPException(
|
||
status_code=status.HTTP_307_TEMPORARY_REDIRECT,
|
||
headers={"Location": login_url}
|
||
)
|
||
|
||
def get_current_user(credentials: HTTPBasicCredentials = Depends(security)) -> Optional[dict]:
|
||
"""Get current authenticated user (HTTP Basic Auth fallback for API)"""
|
||
try:
|
||
return verify_credentials(credentials)
|
||
except HTTPException:
|
||
return None
|
||
|
||
@app.get("/login", response_class=HTMLResponse, name="login")
|
||
async def login_page(request: Request, next: str = "/"):
|
||
"""Trang đăng nhập — nếu đã login rồi thì redirect ngay"""
|
||
# Nếu đã đăng nhập thì không cần vào trang login nữa
|
||
if get_session_user(request):
|
||
return RedirectResponse(url=next, status_code=303)
|
||
return templates.TemplateResponse(
|
||
"login.html",
|
||
{"request": request, "message": None, "sso_enabled": load_sso_config().get('enabled', False)}
|
||
)
|
||
|
||
@app.post("/login", name="login_post")
|
||
async def login(
|
||
request: Request,
|
||
username: str = Form(...),
|
||
password: str = Form(...),
|
||
otp: str = Form("")
|
||
):
|
||
"""Xử lý đăng nhập qua RADIUS + OTP 2FA"""
|
||
user_info = None
|
||
error_message = "Tên đăng nhập hoặc mật khẩu không đúng"
|
||
|
||
# Form gửi lên email dạng ten.nv@vnpt.vn → tách phần trước @ làm username RADIUS
|
||
email_input = username.strip().lower()
|
||
if '@' in email_input:
|
||
username = email_input.split('@')[0] # "ten.nv"
|
||
else:
|
||
username = email_input
|
||
full_email = email_input if '@' in email_input else f"{email_input}@vnpt.vn"
|
||
|
||
# Ghép password + OTP để gửi tới RADIUS
|
||
radius_password = f"{password}{otp}" if otp else password
|
||
|
||
# Xác thực qua RADIUS (Chạy trong thread pool để không block main event loop)
|
||
if RADIUS_CONFIG.get('enabled', False):
|
||
import asyncio
|
||
radius_success, radius_message, radius_attrs = await asyncio.to_thread(authenticate_radius, username, radius_password)
|
||
if radius_success:
|
||
if username in AUTHORIZED_USERS:
|
||
# User có trong whitelist → lấy thông tin đầy đủ (admin hoặc role được cấu hình)
|
||
user_info = dict(AUTHORIZED_USERS[username])
|
||
user_info['auth_method'] = 'radius'
|
||
|
||
# Ưu tiên cao nhất: fixed_unit_override (ghi đè LDAP/SSO)
|
||
override_name, override_id = get_fixed_unit_override(username)
|
||
if override_id:
|
||
user_info['unit_id'] = override_id
|
||
user_info['unit_name'] = override_name
|
||
else:
|
||
# Fallback: Cập nhật thông tin đơn vị từ unit_mapping.json
|
||
m_name, m_id = get_unit_info_from_username(username)
|
||
if m_id:
|
||
user_info['unit_id'] = m_id
|
||
user_info['unit_name'] = m_name
|
||
else:
|
||
mapping = _load_unit_mapping()
|
||
units = mapping.get('units', {})
|
||
uid = str(user_info.get('unit_id', ''))
|
||
if uid in units:
|
||
user_info['unit_name'] = units[uid]
|
||
|
||
print(Fore.GREEN + f"[LOGIN] RADIUS OK — whitelist user '{username}' (role: {user_info.get('role', 'user')})" + Fore.RESET)
|
||
else:
|
||
unit_id = ""
|
||
unit_name = ""
|
||
|
||
# 0. Ưu tiên cao nhất: fixed_unit_override (ghi đè hoàn toàn LDAP/SSO)
|
||
unit_name, unit_id = get_fixed_unit_override(username)
|
||
if unit_name or unit_id:
|
||
print(Fore.MAGENTA + f"[LOGIN] Sử dụng đơn vị cố định (override) cho '{username}'" + Fore.RESET)
|
||
else:
|
||
# 1. Thử lấy từ unit_mapping.json theo username trước
|
||
unit_name, unit_id = get_unit_info_from_username(username)
|
||
if unit_name or unit_id:
|
||
print(Fore.CYAN + f"[LOGIN] Unit từ users mapping: '{unit_name}' (ID: {unit_id})" + Fore.RESET)
|
||
else:
|
||
# 2. Nếu không có mapping trực tiếp, lấy danh sách groups của user từ RADIUS hoặc privacyIDEA
|
||
groups = radius_attrs.get('filter_id', [])
|
||
if groups:
|
||
print(Fore.CYAN + f"[LOGIN] Tìm thấy {len(groups)} group(s) từ RADIUS Filter-Id" + Fore.RESET)
|
||
else:
|
||
pi_cfg = RADIUS_CONFIG.get('privacyidea', {})
|
||
if pi_cfg.get('enabled') and pi_cfg.get('admin_password'):
|
||
groups, _ = await asyncio.to_thread(get_groups_from_privacyidea, username)
|
||
if groups:
|
||
print(Fore.CYAN + f"[LOGIN] Tìm thấy {len(groups)} group(s) từ privacyIDEA API" + Fore.RESET)
|
||
|
||
# 3. Map groups -> unit_id, unit_name từ unit_mapping.json
|
||
if groups:
|
||
unit_name, unit_id = map_groups_to_unit(groups)
|
||
if unit_name or unit_id:
|
||
print(Fore.CYAN + f"[LOGIN] Đã map groups -> unit_name: '{unit_name}', unit_id: {unit_id}" + Fore.RESET)
|
||
else:
|
||
print(Fore.YELLOW + f"[LOGIN] Các group của user không khớp với 'group_mapping' trong unit_mapping.json" + Fore.RESET)
|
||
|
||
if not unit_name and not unit_id:
|
||
print(Fore.YELLOW + f"[LOGIN] Không tìm thấy đơn vị cho '{username}' → unit_name và unit_id để trống" + Fore.RESET)
|
||
|
||
# Phân quyền: kiểm tra user có nằm trong danh sách admins không
|
||
is_admin = is_admin_in_mapping(username)
|
||
role = "admin" if is_admin else "user"
|
||
|
||
user_info = {
|
||
"username": username,
|
||
"full_name": username,
|
||
"email": full_email,
|
||
"role": role,
|
||
"unit_id": unit_id,
|
||
"unit_name": unit_name,
|
||
"auth_method": "radius",
|
||
"can_view_outputs": True,
|
||
"can_download": True,
|
||
}
|
||
print(Fore.CYAN + f"[LOGIN] RADIUS OK — '{username}' ({full_email}) → unit='{unit_name or 'N/A'}', role: {role}" + Fore.RESET)
|
||
else:
|
||
# RADIUS xác thực thất bại
|
||
if "timeout" in radius_message.lower() or "error" in radius_message.lower():
|
||
error_message = "Không thể kết nối tới máy chủ xác thực. Vui lòng thử lại sau."
|
||
elif otp:
|
||
error_message = "Sai tên đăng nhập, mật khẩu hoặc mã OTP"
|
||
else:
|
||
error_message = "Sai thông tin đăng nhập hoặc thiếu mã OTP"
|
||
else:
|
||
error_message = "Hệ thống xác thực RADIUS chưa được cấu hình"
|
||
|
||
# Fallback: chỉ cho phép admin đăng nhập local khi RADIUS không khả dụng + IP hợp lệ
|
||
if not user_info:
|
||
local_success, user = authenticate_local(username, password)
|
||
if local_success:
|
||
# Kiểm tra IP client có trong danh sách cho phép không (nếu là admin)
|
||
client_ip = request.headers.get('X-Forwarded-For', '').split(',')[0].strip() or request.client.host
|
||
ip_allowed = True
|
||
|
||
if user.get('role') == 'admin':
|
||
allowed_ips = RADIUS_CONFIG.get('admin_fallback_allowed_ips', [])
|
||
if allowed_ips:
|
||
ip_allowed = False
|
||
try:
|
||
client_addr = ipaddress.ip_address(client_ip)
|
||
for allowed in allowed_ips:
|
||
try:
|
||
if '/' in allowed:
|
||
if client_addr in ipaddress.ip_network(allowed, strict=False):
|
||
ip_allowed = True
|
||
break
|
||
else:
|
||
if client_addr == ipaddress.ip_address(allowed):
|
||
ip_allowed = True
|
||
break
|
||
except ValueError:
|
||
continue
|
||
except ValueError:
|
||
print(Fore.RED + f"[LOGIN] Invalid client IP: {client_ip}" + Fore.RESET)
|
||
|
||
if ip_allowed:
|
||
user_info = user
|
||
print(Fore.YELLOW + f"[LOGIN] Local login for: {username} from IP: {client_ip}" + Fore.RESET)
|
||
else:
|
||
print(Fore.RED + f"[LOGIN] Admin fallback DENIED for: {username} from IP: {client_ip} (not in whitelist)" + Fore.RESET)
|
||
error_message = f"Địa chỉ IP {client_ip} không được phép đăng nhập admin"
|
||
|
||
if user_info:
|
||
# Lấy hoặc tạo session
|
||
session_id = get_or_create_session(request)
|
||
sessions[session_id]['username'] = user_info.get('username', username)
|
||
sessions[session_id]['user_info'] = user_info
|
||
# Lưu unit_id vào session để gắn vào nội dung upload
|
||
sessions[session_id]['unit_id'] = str(user_info.get('unit_id', ''))
|
||
sessions[session_id]['unit_name'] = user_info.get('unit_name', '')
|
||
sessions[session_id]['fullname'] = user_info.get('full_name', username)
|
||
sessions[session_id]['sso_authenticated'] = False # Đăng nhập thường
|
||
|
||
print(Fore.GREEN + f"[LOGIN] User '{username}' logged in (method: {user_info.get('auth_method', 'local')}, unit_id: {user_info.get('unit_id', 'N/A')})" + Fore.RESET)
|
||
|
||
# Redirect về next URL hoặc trang chủ
|
||
next_url = request.query_params.get('next', str(request.url_for('index')))
|
||
# Bảo vệ open redirect: chỉ cho phép redirect nội bộ
|
||
if next_url and not next_url.startswith('/'):
|
||
next_url = str(request.url_for('index'))
|
||
response = RedirectResponse(url=next_url, status_code=303)
|
||
response.set_cookie(
|
||
key="session_id",
|
||
value=session_id,
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600,
|
||
httponly=True,
|
||
samesite='lax'
|
||
)
|
||
# Lưu username vào cookie riêng để phục hồi session sau khi server restart
|
||
response.set_cookie(
|
||
key="logged_in_user",
|
||
value=user_info.get('username', username),
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600,
|
||
httponly=True,
|
||
samesite='lax'
|
||
)
|
||
# Lưu unit_id, unit_name để gắn vào file khi upload (dùng khi session bị mất)
|
||
response.set_cookie(
|
||
key="user_unit_id",
|
||
value=str(user_info.get('unit_id', '')),
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600,
|
||
httponly=True,
|
||
samesite='lax'
|
||
)
|
||
response.set_cookie(
|
||
key="user_unit_name",
|
||
value=str(user_info.get('unit_name', '')),
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600,
|
||
httponly=True,
|
||
samesite='lax'
|
||
)
|
||
response.set_cookie(
|
||
key="user_role",
|
||
value=str(user_info.get('role', 'user')),
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600,
|
||
httponly=True,
|
||
samesite='lax'
|
||
)
|
||
return response
|
||
|
||
# Đăng nhập thất bại
|
||
return templates.TemplateResponse(
|
||
"login.html",
|
||
{
|
||
"request": request,
|
||
"message": {"type": "error", "text": error_message},
|
||
"sso_enabled": load_sso_config().get('enabled', False)
|
||
}
|
||
)
|
||
|
||
@app.get("/logout", name="logout")
|
||
async def logout(request: Request):
|
||
"""Đăng xuất - xóa thông tin user khỏi session"""
|
||
session_id = request.cookies.get('session_id')
|
||
|
||
if session_id and session_id in sessions:
|
||
# Xóa thông tin user nhưng giữ lại session và files
|
||
sessions[session_id]['username'] = None
|
||
sessions[session_id]['user_info'] = None
|
||
print(Fore.YELLOW + f"[LOGOUT] User logged out from session: {session_id[:8]}..." + Fore.RESET)
|
||
|
||
# Xóa tất cả cookie xác thực khi logout
|
||
# Redirect về trang chủ
|
||
response = RedirectResponse(url=request.url_for('index'), status_code=303)
|
||
response.delete_cookie(key="logged_in_user")
|
||
response.delete_cookie(key="user_unit_id")
|
||
response.delete_cookie(key="user_unit_name")
|
||
response.delete_cookie(key="user_role")
|
||
return response
|
||
|
||
@app.get("/", response_class=HTMLResponse, name="index")
|
||
async def index(request: Request):
|
||
# Kiểm tra đăng nhập bắt buộc
|
||
session_id = get_or_create_session(request)
|
||
user_info = sessions.get(session_id, {}).get('user_info')
|
||
|
||
if not user_info:
|
||
# Chưa đăng nhập → redirect về trang login
|
||
response = RedirectResponse(url=str(request.url_for('login')), status_code=303)
|
||
response.set_cookie(
|
||
key="session_id",
|
||
value=session_id,
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600,
|
||
httponly=True,
|
||
samesite='lax'
|
||
)
|
||
return response
|
||
|
||
is_logged_in = True
|
||
|
||
# Lấy root_path từ ASGI scope hoặc proxy headers
|
||
root_path = get_app_root_path(request)
|
||
|
||
# Lấy email người dùng đã đăng ký trong session (nếu có)
|
||
user_email = sessions.get(session_id, {}).get('user_email', '')
|
||
|
||
html_response = templates.TemplateResponse(
|
||
"index.html",
|
||
{
|
||
"request": request,
|
||
"os_config": OS_CONFIG,
|
||
"message": None,
|
||
"is_logged_in": is_logged_in,
|
||
"user_info": user_info,
|
||
"tools_list": get_available_tools(),
|
||
"root_path": root_path,
|
||
"user_email": user_email,
|
||
}
|
||
)
|
||
|
||
# Prevent browser caching
|
||
html_response.headers["Cache-Control"] = "no-cache, no-store, must-revalidate"
|
||
html_response.headers["Pragma"] = "no-cache"
|
||
html_response.headers["Expires"] = "0"
|
||
|
||
# Set session cookie nếu chưa có
|
||
if not request.cookies.get('session_id'):
|
||
html_response.set_cookie(
|
||
key="session_id",
|
||
value=session_id,
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600,
|
||
httponly=True,
|
||
samesite='lax'
|
||
)
|
||
|
||
return html_response
|
||
|
||
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
# SSO Portal Authentication Endpoint
|
||
# Áp dụng theo luồng 8 bước từ portal_auth_flow_documentation.md
|
||
# ─────────────────────────────────────────────────────────────────────────────
|
||
@app.get("/portal_auth", name="portal_auth")
|
||
async def portal_auth(
|
||
request: Request,
|
||
jwt: Optional[str] = None,
|
||
next: Optional[str] = None,
|
||
):
|
||
"""
|
||
Entry point SSO authentication.
|
||
Nhận JWT token từ SSO Portal qua query params và xác thực với SSO Validate Server.
|
||
URL: GET /portal_auth?jwt=<token>&next=<redirect_path>
|
||
"""
|
||
# ── Bước 0: Kiểm tra xem user đã authenticated chưa ──────────────────────
|
||
session_id = get_or_create_session(request)
|
||
existing_user = sessions.get(session_id, {}).get('user_info')
|
||
if existing_user:
|
||
# Đã đăng nhập → redirect luôn, bỏ qua toàn bộ SSO flow
|
||
redirect_url = _safe_redirect(next, default=request.scope.get("root_path", "") + "/")
|
||
print(Fore.CYAN + f"[SSO] User already authenticated: {existing_user.get('username')} → redirect {redirect_url}" + Fore.RESET)
|
||
return RedirectResponse(url=redirect_url, status_code=303)
|
||
|
||
# ── Bước 1: Kiểm tra JWT token có trong query params ─────────────────────
|
||
if not jwt:
|
||
print(Fore.RED + "[SSO] Missing JWT token in query params" + Fore.RESET)
|
||
raise HTTPException(status_code=400, detail="Missing JWT token")
|
||
|
||
sso_cfg = load_sso_config()
|
||
if not sso_cfg.get('enabled', False):
|
||
raise HTTPException(status_code=503, detail="SSO authentication is not enabled on this server")
|
||
|
||
# ── Bước 2: Validate next redirect URL (chống Open Redirect) ─────────────
|
||
redirect_url = _safe_redirect(next, default=request.scope.get("root_path", "") + "/")
|
||
|
||
# ── Bước 3: Gọi SSO Validate Token API ───────────────────────────────────
|
||
sso_validate_url = sso_cfg.get('validate_url')
|
||
sso_timeout = sso_cfg.get('timeout', 10)
|
||
sso_verify_ssl = sso_cfg.get('verify_ssl', False)
|
||
|
||
print(Fore.CYAN + f"[SSO] Validating JWT token via: {sso_validate_url}" + Fore.RESET)
|
||
|
||
try:
|
||
resp = http_requests.get(
|
||
sso_validate_url,
|
||
headers={"Authorization": f"Bearer {jwt}"},
|
||
timeout=sso_timeout,
|
||
verify=sso_verify_ssl
|
||
)
|
||
except http_requests.exceptions.RequestException as e:
|
||
print(Fore.RED + f"[SSO] SSO server unreachable: {e}" + Fore.RESET)
|
||
raise HTTPException(status_code=503, detail=f"SSO service unreachable: {str(e)}")
|
||
|
||
# ── Bước 4: Kiểm tra kết quả validate ────────────────────────────────────
|
||
if resp.status_code != 200:
|
||
print(Fore.RED + f"[SSO] SSO server returned HTTP {resp.status_code}" + Fore.RESET)
|
||
raise HTTPException(status_code=401, detail="Invalid or expired token")
|
||
|
||
try:
|
||
payload = resp.json()
|
||
except Exception:
|
||
raise HTTPException(status_code=502, detail="Invalid SSO response format")
|
||
|
||
if payload.get("message") != "success":
|
||
print(Fore.RED + f"[SSO] SSO message not success: {payload.get('message')}" + Fore.RESET)
|
||
raise HTTPException(status_code=401, detail="SSO authentication failed")
|
||
|
||
data = payload.get("data", {})
|
||
|
||
# In toàn bộ tham số trả về để kiểm tra
|
||
print(Fore.YELLOW + f"[SSO] Raw data payload: {data}" + Fore.RESET)
|
||
|
||
# Kiểm tra token hết hạn
|
||
import time as _time
|
||
exp = data.get("exp")
|
||
if exp and exp < int(_time.time()):
|
||
print(Fore.RED + "[SSO] Token expired" + Fore.RESET)
|
||
raise HTTPException(status_code=401, detail="Token expired")
|
||
|
||
# Kiểm tra email
|
||
email = data.get("email", "").strip()
|
||
if not email:
|
||
raise HTTPException(status_code=401, detail="Missing email in token")
|
||
|
||
# ── Bước 5: Tách username từ email ───────────────────────────────────────
|
||
username = email.split('@')[0]
|
||
fullname = data.get("fullname", username)
|
||
sso_unit_id = data.get("unit_id", 0)
|
||
|
||
print(Fore.GREEN + f"[SSO] Token valid for: {username} (unit_id={sso_unit_id}, fullname={fullname})" + Fore.RESET)
|
||
|
||
# ── Bước 6: Tạo user_info từ SSO payload (không cần trong whitelist) ─────
|
||
# Xác định role cho user SSO:
|
||
# 1. Ưu tiên lấy role từ payload SSO nếu có
|
||
sso_role = data.get("role") or data.get("role_name")
|
||
|
||
# 2. Map role theo unit_id từ SSO (Nếu unit_id == 1 -> admin)
|
||
if not sso_role:
|
||
if str(sso_unit_id) == "1":
|
||
sso_role = "admin"
|
||
else:
|
||
# 3. Fallback theo file cấu hình nội bộ của Tool
|
||
is_admin = is_admin_in_mapping(username)
|
||
sso_role = "admin" if is_admin else "user"
|
||
|
||
# Ánh xạ tên đơn vị (unit_name) từ unit_id
|
||
# Ưu tiên cao nhất: fixed_unit_override (ghi đè hoàn toàn đơn vị từ SSO)
|
||
override_name, override_id = get_fixed_unit_override(username)
|
||
if override_id:
|
||
sso_unit_id = override_id
|
||
sso_unit_name = override_name
|
||
print(Fore.MAGENTA + f"[SSO] Sử dụng đơn vị cố định (override) cho '{username}': '{override_name}' (ID: {override_id})" + Fore.RESET)
|
||
else:
|
||
mapping = _load_unit_mapping()
|
||
units = mapping.get('units', {})
|
||
|
||
if not sso_unit_id or str(sso_unit_id) == "0":
|
||
m_name, m_id = get_unit_info_from_username(username)
|
||
if m_id:
|
||
sso_unit_id = m_id
|
||
sso_unit_name = m_name
|
||
else:
|
||
sso_unit_name = data.get("unit_name", "Unknown Unit")
|
||
else:
|
||
sso_unit_name = units.get(str(sso_unit_id))
|
||
if not sso_unit_name:
|
||
sso_unit_name = data.get("unit_name", f"Unit {sso_unit_id}")
|
||
|
||
sso_user_info = {
|
||
"username": username,
|
||
"full_name": fullname,
|
||
"email": email,
|
||
"role": sso_role,
|
||
"unit_id": sso_unit_id,
|
||
"unit_name": sso_unit_name,
|
||
"auth_method": "sso",
|
||
"can_view_outputs": True,
|
||
"can_download": True,
|
||
}
|
||
|
||
# ── Bước 7: Tạo session và lưu thông tin ─────────────────────────────────
|
||
sessions[session_id]['username'] = username
|
||
sessions[session_id]['user_info'] = sso_user_info
|
||
sessions[session_id]['unit_id'] = str(sso_unit_id)
|
||
sessions[session_id]['unit_name'] = sso_user_info['unit_name']
|
||
sessions[session_id]['fullname'] = fullname
|
||
sessions[session_id]['sso_authenticated'] = True
|
||
|
||
# Ghi log đăng nhập
|
||
client_ip = request.headers.get('X-Forwarded-For', '').split(',')[0].strip() or request.client.host
|
||
ua = request.headers.get('user-agent', '')
|
||
print(Fore.GREEN + f"[SSO] ✓ User '{username}' authenticated via SSO Portal — IP: {client_ip}, UA: {ua[:60]}" + Fore.RESET)
|
||
|
||
# ── Bước 8: Redirect cuối ─────────────────────────────────────────────────
|
||
response = RedirectResponse(url=redirect_url, status_code=303)
|
||
response.set_cookie(
|
||
key="session_id",
|
||
value=session_id,
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600,
|
||
httponly=True,
|
||
samesite='lax'
|
||
)
|
||
# Lưu thông tin user vào cookie để phục hồi session sau khi server restart
|
||
response.set_cookie(key="logged_in_user", value=username,
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600, httponly=True, samesite='lax')
|
||
response.set_cookie(key="user_unit_id", value=str(sso_unit_id),
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600, httponly=True, samesite='lax')
|
||
response.set_cookie(key="user_unit_name", value=sso_user_info['unit_name'],
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600, httponly=True, samesite='lax')
|
||
response.set_cookie(key="user_role", value=sso_role,
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600, httponly=True, samesite='lax')
|
||
return response
|
||
|
||
|
||
def _safe_redirect(next_url: Optional[str], default: str = "/") -> str:
|
||
"""Kiểm tra và trả về redirect URL an toàn (chỉ cho phép path nội bộ, chống Open Redirect)."""
|
||
if not next_url:
|
||
return default
|
||
# Chỉ chấp nhận relative path (bắt đầu bằng /)
|
||
if next_url.startswith('/'):
|
||
return next_url
|
||
# Từ chối URL bên ngoài
|
||
return default
|
||
|
||
@app.post("/api/register_email")
|
||
async def api_register_email(request: Request, email: str = Form(...)):
|
||
"""
|
||
Nhận email người dùng (@vnpt.vn), validate và lưu vào session.
|
||
Trả về JSON {success, message}.
|
||
"""
|
||
email = email.strip().lower()
|
||
|
||
if not validate_vnpt_email(email):
|
||
return {
|
||
"success": False,
|
||
"message": "Email không hợp lệ. Vui lòng nhập địa chỉ email dạng @vnpt.vn."
|
||
}
|
||
|
||
session_id = get_or_create_session(request)
|
||
if session_id in sessions:
|
||
sessions[session_id]['user_email'] = email
|
||
print(Fore.CYAN + f"[EMAIL-REG] Registered email '{email}' for session {session_id[:8]}..." + Fore.RESET)
|
||
|
||
# Tạo/cập nhật bản ghi trong DB nếu chưa có (first_seen)
|
||
db = SessionLocal()
|
||
try:
|
||
existing = db.query(UserEmailRecord).filter(UserEmailRecord.email == email).first()
|
||
if not existing:
|
||
record = UserEmailRecord(
|
||
email=email,
|
||
display_name=email.split('@')[0],
|
||
total_uploads=0,
|
||
total_files=0,
|
||
)
|
||
db.add(record)
|
||
db.commit()
|
||
except Exception as e:
|
||
db.rollback()
|
||
print(Fore.RED + f"[EMAIL-REG] DB error: {e}" + Fore.RESET)
|
||
finally:
|
||
db.close()
|
||
|
||
return {"success": True, "message": f"Đã xác nhận email: {email}"}
|
||
|
||
@app.post("/api/detect_os")
|
||
async def api_detect_os(filenames: List[str] = Form(...)):
|
||
"""
|
||
API endpoint to detect OS from filenames.
|
||
Returns detected OS type and whether detection was successful.
|
||
"""
|
||
try:
|
||
detected_os, all_same = detect_os_from_multiple_files(filenames)
|
||
|
||
if detected_os:
|
||
os_info = OS_CONFIG.get(detected_os, {})
|
||
return {
|
||
"success": True,
|
||
"detected_os": detected_os,
|
||
"os_name": os_info.get('name', detected_os),
|
||
"all_same": all_same,
|
||
"message": f"Tự động phát hiện: {os_info.get('name', detected_os)}" if all_same
|
||
else f"Các file có OS khác nhau. Đề xuất: {os_info.get('name', detected_os)}"
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"detected_os": None,
|
||
"os_name": None,
|
||
"all_same": True,
|
||
"message": "Không thể tự động phát hiện OS. Vui lòng chọn thủ công."
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"detected_os": None,
|
||
"os_name": None,
|
||
"all_same": True,
|
||
"message": f"Error: {str(e)}"
|
||
}
|
||
|
||
@app.post("/api/detect_os_from_content")
|
||
async def api_detect_os_from_content(files: List[UploadFile] = File(...)):
|
||
"""
|
||
API endpoint to detect OS from file contents.
|
||
It decrypts the first valid file and extracts the OS information.
|
||
"""
|
||
try:
|
||
from werkzeug.utils import secure_filename
|
||
import shutil
|
||
print(Fore.CYAN + "[API] Auto-detecting OS from file content..." + Fore.RESET)
|
||
|
||
detected_os = None
|
||
for f in files:
|
||
if f.filename and allowed_file(f.filename):
|
||
_temp_fname = "api_detect_" + secure_filename(f.filename)
|
||
_temp_enc_path = os.path.join(UPLOAD_FOLDER, _temp_fname)
|
||
_original_dir = os.getcwd()
|
||
try:
|
||
_content_bytes = await f.read()
|
||
with open(_temp_enc_path, "wb") as _buf:
|
||
_buf.write(_content_bytes)
|
||
detected_os = detect_os_from_decrypted_file(_temp_enc_path)
|
||
except Exception as _e:
|
||
print(Fore.YELLOW + f"[API-DETECT] Content detection error: {_e}" + Fore.RESET)
|
||
finally:
|
||
os.chdir(_original_dir)
|
||
# Clean up temp file
|
||
if os.path.exists(_temp_enc_path):
|
||
os.remove(_temp_enc_path)
|
||
|
||
if detected_os:
|
||
break
|
||
|
||
if detected_os:
|
||
os_info = OS_CONFIG.get(detected_os, {})
|
||
return {
|
||
"success": True,
|
||
"detected_os": detected_os,
|
||
"os_name": os_info.get('name', detected_os),
|
||
"message": f"Phân tích nội dung file: {os_info.get('name', detected_os)}"
|
||
}
|
||
else:
|
||
return {
|
||
"success": False,
|
||
"detected_os": None,
|
||
"os_name": None,
|
||
"message": "Không thể tự động phát hiện OS từ nội dung."
|
||
}
|
||
except Exception as e:
|
||
return {
|
||
"success": False,
|
||
"detected_os": None,
|
||
"os_name": None,
|
||
"message": f"Error: {str(e)}"
|
||
}
|
||
|
||
def process_files_background(task_id: str, session_id: str, os_type: str, enc_paths: List[str], base_url: str):
|
||
"""Background task to process uploaded files"""
|
||
try:
|
||
tasks_status = processing_tasks.get(task_id)
|
||
if not tasks_status:
|
||
return
|
||
|
||
tasks_status['status'] = 'processing'
|
||
tasks_status['start_time'] = time.time()
|
||
|
||
# Validation checks
|
||
if os_type != 'auto' and os_type not in OS_CONFIG:
|
||
tasks_status['status'] = 'error'
|
||
tasks_status['message'] = "Please select a valid OS type!"
|
||
return
|
||
|
||
if not enc_paths:
|
||
tasks_status['status'] = 'error'
|
||
tasks_status['message'] = "No encrypted files found!"
|
||
return
|
||
|
||
output_files = []
|
||
system_info_list = []
|
||
|
||
# User auth check
|
||
is_admin = False
|
||
if session_id in sessions:
|
||
user_info = sessions[session_id].get('user_info')
|
||
if user_info and user_info.get('role') == 'admin':
|
||
is_admin = True
|
||
|
||
total_files = len(enc_paths)
|
||
for i, enc_path in enumerate(enc_paths):
|
||
enc_filename = os.path.basename(enc_path)
|
||
tasks_status['progress'] = int(10 + (i / total_files) * 80)
|
||
tasks_status['message'] = f"Đang xử lý {i+1}/{total_files}: {enc_filename}"
|
||
|
||
# Determine OS type per file if os_type == 'auto'
|
||
file_os_type = os_type
|
||
if file_os_type == 'auto':
|
||
file_os_type = detect_os_from_filename(enc_filename)
|
||
if not file_os_type:
|
||
# Fallback to content detection
|
||
file_os_type = detect_os_from_decrypted_file(enc_path)
|
||
|
||
if not file_os_type or file_os_type not in OS_CONFIG:
|
||
print(Fore.YELLOW + f"[BACKGROUND] Cannot detect OS for {enc_filename}, skipping." + Fore.RESET)
|
||
continue
|
||
|
||
os_info = OS_CONFIG[file_os_type]
|
||
checklist_path = os.path.join(CONFIG_FOLDER, os_info['checklist'])
|
||
config_path = os.path.join(CONFIG_FOLDER, os_info['config'])
|
||
|
||
if not os.path.exists(checklist_path) or not os.path.exists(config_path):
|
||
print(Fore.RED + f"[BACKGROUND] Missing checklist/config for {file_os_type}, skipping." + Fore.RESET)
|
||
continue
|
||
|
||
print(Fore.BLUE + f"\n=== [BACKGROUND] Processing: {enc_filename} ({os_info['name']}) ===" + Fore.RESET)
|
||
original_dir = os.getcwd()
|
||
|
||
try:
|
||
system_info = audit_decryption.run_generate_excel(checklist_path, config_path, enc_path)
|
||
|
||
dec_filename = enc_filename.replace('.enc', '')
|
||
if os.path.exists(dec_filename):
|
||
output_dec_path = os.path.join(OUTPUT_FOLDER, dec_filename)
|
||
if os.path.exists(output_dec_path):
|
||
os.remove(output_dec_path)
|
||
shutil.move(dec_filename, output_dec_path)
|
||
|
||
search_dirs = ['.', UPLOAD_FOLDER, os.path.dirname(__file__)]
|
||
excel_found = False
|
||
|
||
for search_dir in search_dirs:
|
||
if not os.path.exists(search_dir):
|
||
continue
|
||
for file in os.listdir(search_dir):
|
||
if file.endswith('.xlsx') and 'Checklist' not in file:
|
||
source_path = os.path.join(search_dir, file)
|
||
output_path = os.path.join(OUTPUT_FOLDER, file)
|
||
if os.path.exists(output_path):
|
||
os.remove(output_path)
|
||
shutil.move(source_path, output_path)
|
||
output_files.append(file)
|
||
excel_found = True
|
||
print(Fore.GREEN + f"[OK] Generated: {file}" + Fore.RESET)
|
||
break
|
||
if excel_found:
|
||
break
|
||
|
||
if system_info:
|
||
system_info_list.append(system_info)
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
print(Fore.RED + f"Error processing {enc_filename}: {str(e)}" + Fore.RESET)
|
||
print(traceback.format_exc())
|
||
finally:
|
||
os.chdir(original_dir)
|
||
|
||
if output_files:
|
||
tasks_status['progress'] = 90
|
||
tasks_status['message'] = "Đang hoàn tất và đồng bộ dữ liệu..."
|
||
|
||
for filename in output_files:
|
||
add_file_to_session(session_id, filename)
|
||
|
||
try:
|
||
extract_info_from_output_files()
|
||
except Exception as e:
|
||
print(Fore.RED + f"[DB SYNC ERROR] {e}" + Fore.RESET)
|
||
|
||
# Ghi nhận thống kê upload theo email người dùng (nếu có)
|
||
user_email_for_stat = sessions.get(session_id, {}).get('user_email', '')
|
||
record_email_upload(user_email_for_stat, len(output_files))
|
||
|
||
email_sent = False
|
||
if _email_enabled():
|
||
tasks_status['message'] = "Đang gửi báo cáo qua Email..."
|
||
try:
|
||
email_attachments = []
|
||
for filename in output_files:
|
||
excel_path = os.path.join(OUTPUT_FOLDER, filename)
|
||
if os.path.exists(excel_path):
|
||
email_attachments.append(excel_path)
|
||
|
||
for enc_path in enc_paths:
|
||
dec_filename = os.path.basename(enc_path).replace('.enc', '')
|
||
txt_path = os.path.join(OUTPUT_FOLDER, dec_filename)
|
||
if os.path.exists(txt_path):
|
||
email_attachments.append(txt_path)
|
||
|
||
processing_time = time.time() - tasks_status['start_time']
|
||
send_processing_complete_email(
|
||
output_files=output_files,
|
||
os_type=os_type,
|
||
processing_time=processing_time,
|
||
base_url=base_url,
|
||
system_info_list=system_info_list,
|
||
attachments=email_attachments,
|
||
is_admin=True
|
||
)
|
||
email_sent = True
|
||
print(Fore.GREEN + "[EMAIL] Notification sent successfully" + Fore.RESET)
|
||
except Exception as e:
|
||
print(Fore.YELLOW + f"[EMAIL] Failed to send notification: {e}" + Fore.RESET)
|
||
|
||
decrypted_files = []
|
||
if is_admin:
|
||
for enc_path in enc_paths:
|
||
dec_filename = os.path.basename(enc_path).replace('.enc', '')
|
||
txt_path = os.path.join(OUTPUT_FOLDER, dec_filename)
|
||
if os.path.exists(txt_path):
|
||
decrypted_files.append(dec_filename)
|
||
|
||
# Tạo message phản ánh đúng thực tế việc gửi mail
|
||
if email_sent:
|
||
result_msg_text = f"Đã xử lý thành công {len(output_files)} file. ✉️ Đã gửi email thông báo."
|
||
elif _email_enabled():
|
||
result_msg_text = f"Đã xử lý thành công {len(output_files)} file. ⚠️ Gửi email thất bại."
|
||
else:
|
||
result_msg_text = f"Đã xử lý thành công {len(output_files)} file."
|
||
|
||
tasks_status['result'] = {
|
||
"output_files": output_files,
|
||
"decrypted_files": decrypted_files,
|
||
"is_admin": is_admin,
|
||
"system_info_list": system_info_list,
|
||
"email_sent": email_sent,
|
||
"message": {"type": "success", "text": result_msg_text}
|
||
}
|
||
tasks_status['progress'] = 100
|
||
tasks_status['status'] = 'completed'
|
||
tasks_status['message'] = "Hoàn tất!"
|
||
else:
|
||
tasks_status['status'] = 'error'
|
||
tasks_status['message'] = "Processing completed but no output files were generated."
|
||
except Exception as e:
|
||
tasks_status['status'] = 'error'
|
||
tasks_status['message'] = f"Lỗi hệ thống: {str(e)}"
|
||
|
||
@app.post("/api/upload_async")
|
||
async def api_upload_async(
|
||
request: Request,
|
||
background_tasks: BackgroundTasks,
|
||
os_type: str = Form(...),
|
||
encrypted_files: List[UploadFile] = File(...)
|
||
):
|
||
# Bắt buộc đăng nhập để xác định unit_id người upload
|
||
session_id = get_or_create_session(request)
|
||
user_info = sessions.get(session_id, {}).get('user_info')
|
||
if not user_info:
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="Vui lòng đăng nhập trước khi tải lên file"
|
||
)
|
||
task_id = str(uuid.uuid4())
|
||
|
||
# Lấy unit_id từ session để gắn vào kết quả xử lý
|
||
session_unit_id = sessions.get(session_id, {}).get('unit_id', '')
|
||
session_unit_name = sessions.get(session_id, {}).get('unit_name', '')
|
||
session_username = user_info.get('username', '')
|
||
|
||
processing_tasks[task_id] = {
|
||
"status": "pending",
|
||
"progress": 0,
|
||
"message": "Đang chuẩn bị dữ liệu...",
|
||
"result": None,
|
||
"start_time": time.time(),
|
||
"unit_id": session_unit_id,
|
||
"unit_name": session_unit_name,
|
||
"username": session_username,
|
||
}
|
||
|
||
try:
|
||
# Detect OS dynamically from filenames (fast, no decryption needed)
|
||
if not os_type or os_type == 'auto':
|
||
filenames = [f.filename for f in encrypted_files if f.filename]
|
||
detected_os, all_same = detect_os_from_multiple_files(filenames)
|
||
if detected_os and all_same:
|
||
os_type = detected_os
|
||
else:
|
||
# Nếu mixed OS hoặc chưa detect hết qua filename, giữ os_type='auto'
|
||
# để process_files_background tự động giải mã và detect theo từng file
|
||
os_type = 'auto'
|
||
|
||
processing_tasks[task_id]['progress'] = 5
|
||
processing_tasks[task_id]['message'] = "Đang lưu trữ file tạm..."
|
||
|
||
enc_paths = []
|
||
for enc_file in encrypted_files:
|
||
if enc_file.filename and allowed_file(enc_file.filename):
|
||
enc_filename = secure_filename(enc_file.filename)
|
||
enc_path = os.path.join(UPLOAD_FOLDER, enc_filename)
|
||
with open(enc_path, "wb") as buffer:
|
||
content = await enc_file.read()
|
||
buffer.write(content)
|
||
enc_paths.append(enc_path)
|
||
|
||
base_url = f"{request.url.scheme}://{request.url.netloc}"
|
||
background_tasks.add_task(process_files_background, task_id, session_id, os_type, enc_paths, base_url)
|
||
|
||
return {"task_id": task_id, "status": "pending"}
|
||
except Exception as e:
|
||
processing_tasks[task_id]['status'] = 'error'
|
||
processing_tasks[task_id]['message'] = f"Lỗi khởi tạo: {str(e)}"
|
||
return {"task_id": task_id, "status": "error"}
|
||
|
||
@app.get("/api/task_status/{task_id}")
|
||
async def task_status(task_id: str):
|
||
task = processing_tasks.get(task_id)
|
||
if not task:
|
||
raise HTTPException(status_code=404, detail="Task not found")
|
||
return {
|
||
"status": task["status"],
|
||
"progress": task.get("progress", 0),
|
||
"message": task.get("message", "")
|
||
}
|
||
|
||
@app.get("/task_result/{task_id}")
|
||
async def task_result(request: Request, task_id: str):
|
||
task = processing_tasks.get(task_id)
|
||
if not task:
|
||
return RedirectResponse(url='/')
|
||
|
||
if task["status"] == "error":
|
||
return templates.TemplateResponse("index.html", {
|
||
"request": request,
|
||
"os_config": OS_CONFIG,
|
||
"message": {"type": "error", "text": task["message"]},
|
||
"is_logged_in": True if get_session_user(request) else False,
|
||
"user_info": get_session_user(request),
|
||
"tools_list": get_available_tools()
|
||
})
|
||
|
||
if task["status"] != "completed" or not task.get("result"):
|
||
return RedirectResponse(url='/')
|
||
|
||
res = task["result"]
|
||
session_user = get_session_user(request)
|
||
html_response = templates.TemplateResponse("result.html", {
|
||
"request": request,
|
||
"output_files": res["output_files"],
|
||
"decrypted_files": res["decrypted_files"],
|
||
"is_admin": res["is_admin"],
|
||
"system_info_list": res["system_info_list"],
|
||
"message": res["message"],
|
||
"is_logged_in": session_user is not None,
|
||
"user_info": session_user,
|
||
"session_mode": False,
|
||
})
|
||
|
||
session_id = get_or_create_session(request)
|
||
html_response.set_cookie(
|
||
key="session_id",
|
||
value=session_id,
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600,
|
||
httponly=True,
|
||
samesite='lax'
|
||
)
|
||
return html_response
|
||
|
||
@app.post("/upload", name="upload")
|
||
async def upload_files(
|
||
request: Request,
|
||
os_type: str = Form(...),
|
||
encrypted_files: List[UploadFile] = File(...)
|
||
):
|
||
# Tạo response object để set cookie
|
||
response = Response()
|
||
|
||
# Lấy hoặc tạo session
|
||
session_id = get_or_create_session(request, response)
|
||
|
||
# Bắt đầu đo thời gian xử lý
|
||
start_time = time.time()
|
||
|
||
try:
|
||
# Lấy thông tin user cho template context (dùng khi cần trả lỗi)
|
||
_user_info_ctx = sessions[session_id].get('user_info') if session_id in sessions else None
|
||
_is_logged_in_ctx = _user_info_ctx is not None
|
||
|
||
# Auto-detect OS from filenames (fast, no decryption needed)
|
||
if not os_type or os_type == 'auto':
|
||
print(Fore.CYAN + "[AUTO-DETECT] Mode: auto – detecting OS from filenames..." + Fore.RESET)
|
||
filenames = [f.filename for f in encrypted_files if f.filename]
|
||
detected_os, all_same = detect_os_from_multiple_files(filenames)
|
||
|
||
if detected_os:
|
||
os_type = detected_os
|
||
print(Fore.GREEN + f"[AUTO-DETECT] ✓ OS detected from filename: {os_type}" + Fore.RESET)
|
||
else:
|
||
return templates.TemplateResponse(
|
||
"index.html",
|
||
{
|
||
"request": request,
|
||
"os_config": OS_CONFIG,
|
||
"is_logged_in": _is_logged_in_ctx,
|
||
"user_info": _user_info_ctx,
|
||
"tools_list": get_available_tools(),
|
||
"message": {"type": "warning", "text": "Không thể tự động phát hiện OS. Vui lòng chọn OS thủ công."}
|
||
}
|
||
)
|
||
|
||
# Validate OS type
|
||
if os_type not in OS_CONFIG:
|
||
return templates.TemplateResponse(
|
||
"index.html",
|
||
{
|
||
"request": request,
|
||
"os_config": OS_CONFIG,
|
||
"message": {"type": "error", "text": "Please select a valid OS type!"}
|
||
}
|
||
)
|
||
|
||
# Validate encrypted files
|
||
if not encrypted_files:
|
||
return templates.TemplateResponse(
|
||
"index.html",
|
||
{
|
||
"request": request,
|
||
"os_config": OS_CONFIG,
|
||
"message": {"type": "error", "text": "Please select at least one encrypted file!"}
|
||
}
|
||
)
|
||
|
||
# Get checklist and config paths
|
||
os_info = OS_CONFIG[os_type]
|
||
checklist_path = os.path.join(CONFIG_FOLDER, os_info['checklist'])
|
||
config_path = os.path.join(CONFIG_FOLDER, os_info['config'])
|
||
|
||
# Verify files exist
|
||
if not os.path.exists(checklist_path):
|
||
return templates.TemplateResponse(
|
||
"index.html",
|
||
{
|
||
"request": request,
|
||
"os_config": OS_CONFIG,
|
||
"message": {"type": "error", "text": f"Checklist file not found: {os_info['checklist']}"}
|
||
}
|
||
)
|
||
|
||
if not os.path.exists(config_path):
|
||
return templates.TemplateResponse(
|
||
"index.html",
|
||
{
|
||
"request": request,
|
||
"os_config": OS_CONFIG,
|
||
"message": {"type": "error", "text": f"Config file not found: {os_info['config']}"}
|
||
}
|
||
)
|
||
|
||
# Process each encrypted file
|
||
output_files = []
|
||
system_info_list = [] # Store system info from each processed file
|
||
for enc_file in encrypted_files:
|
||
if enc_file.filename and allowed_file(enc_file.filename):
|
||
enc_filename = secure_filename(enc_file.filename)
|
||
enc_path = os.path.join(UPLOAD_FOLDER, enc_filename)
|
||
|
||
# Save uploaded file
|
||
with open(enc_path, "wb") as buffer:
|
||
content = await enc_file.read()
|
||
buffer.write(content)
|
||
|
||
# Process the file
|
||
print(Fore.BLUE + f"\n=== Processing: {enc_filename} ({os_info['name']}) ===" + Fore.RESET)
|
||
|
||
# Save current directory
|
||
original_dir = os.getcwd()
|
||
|
||
try:
|
||
# Run decryption and report generation
|
||
system_info = audit_decryption.run_generate_excel(checklist_path, config_path, enc_path)
|
||
|
||
# Find generated decrypted file
|
||
dec_filename = enc_filename.replace('.enc', '')
|
||
if os.path.exists(dec_filename):
|
||
output_dec_path = os.path.join(OUTPUT_FOLDER, dec_filename)
|
||
if os.path.exists(output_dec_path):
|
||
os.remove(output_dec_path)
|
||
shutil.move(dec_filename, output_dec_path)
|
||
print(Fore.CYAN + f"[DEBUG] Moved decrypted file to: {output_dec_path}" + Fore.RESET)
|
||
|
||
# Search for Excel files
|
||
search_dirs = ['.', UPLOAD_FOLDER, os.path.dirname(__file__)]
|
||
excel_found = False
|
||
|
||
for search_dir in search_dirs:
|
||
if not os.path.exists(search_dir):
|
||
continue
|
||
|
||
print(Fore.CYAN + f"[DEBUG] Searching for Excel in: {os.path.abspath(search_dir)}" + Fore.RESET)
|
||
|
||
for file in os.listdir(search_dir):
|
||
if file.endswith('.xlsx') and 'Checklist' not in file:
|
||
source_path = os.path.join(search_dir, file)
|
||
output_path = os.path.join(OUTPUT_FOLDER, file)
|
||
|
||
if os.path.exists(output_path):
|
||
os.remove(output_path)
|
||
|
||
shutil.move(source_path, output_path)
|
||
output_files.append(file)
|
||
excel_found = True
|
||
print(Fore.GREEN + f"[OK] Generated: {file}" + Fore.RESET)
|
||
break
|
||
|
||
if excel_found:
|
||
break
|
||
|
||
if not excel_found:
|
||
print(Fore.YELLOW + f"[WARNING] No Excel file found for {enc_filename}" + Fore.RESET)
|
||
|
||
# Store system info if available
|
||
if system_info:
|
||
system_info_list.append(system_info)
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
print(Fore.RED + f"Error processing {enc_filename}: {str(e)}" + Fore.RESET)
|
||
print(traceback.format_exc())
|
||
|
||
finally:
|
||
os.chdir(original_dir)
|
||
|
||
if output_files:
|
||
# Thêm các file vào session để user có thể download mà không cần xác thực
|
||
for filename in output_files:
|
||
add_file_to_session(session_id, filename)
|
||
|
||
# Cập nhật thông tin vào Database ngay sau khi tạo file
|
||
try:
|
||
import asyncio
|
||
await asyncio.to_thread(extract_info_from_output_files)
|
||
print(Fore.GREEN + "[DB SYNC] Updated database after new file upload." + Fore.RESET)
|
||
except Exception as e:
|
||
print(Fore.RED + f"[DB SYNC ERROR] {e}" + Fore.RESET)
|
||
|
||
# Ghi nhận thống kê upload theo email người dùng (nếu có)
|
||
user_email_for_stat = sessions.get(session_id, {}).get('user_email', '')
|
||
await asyncio.to_thread(record_email_upload, user_email_for_stat, len(output_files))
|
||
|
||
# Tính thời gian xử lý
|
||
processing_time = time.time() - start_time
|
||
|
||
|
||
# Gửi email thông báo
|
||
email_sent = False
|
||
if _email_enabled():
|
||
try:
|
||
# Lấy base URL từ request
|
||
base_url = f"{request.url.scheme}://{request.url.netloc}"
|
||
|
||
# Chuẩn bị danh sách file đính kèm (cả Excel và TXT) - mặc định gửi cho tất cả
|
||
email_attachments = []
|
||
|
||
for filename in output_files:
|
||
excel_path = os.path.join(OUTPUT_FOLDER, filename)
|
||
if os.path.exists(excel_path):
|
||
email_attachments.append(excel_path)
|
||
|
||
# Thêm file TXT đã giải mã
|
||
for enc_file in encrypted_files:
|
||
if enc_file.filename:
|
||
dec_filename = enc_file.filename.replace('.enc', '')
|
||
txt_path = os.path.join(OUTPUT_FOLDER, dec_filename)
|
||
if os.path.exists(txt_path):
|
||
email_attachments.append(txt_path)
|
||
|
||
print(Fore.CYAN + f"[EMAIL] Attaching {len(email_attachments)} file(s) to email" + Fore.RESET)
|
||
|
||
send_processing_complete_email(
|
||
output_files=output_files,
|
||
os_type=os_type,
|
||
processing_time=processing_time,
|
||
base_url=base_url,
|
||
system_info_list=system_info_list,
|
||
attachments=email_attachments,
|
||
is_admin=True # Always attach files
|
||
)
|
||
email_sent = True
|
||
print(Fore.GREEN + "[EMAIL] Notification sent successfully" + Fore.RESET)
|
||
except Exception as email_error:
|
||
print(Fore.YELLOW + f"[EMAIL] Failed to send notification: {email_error}" + Fore.RESET)
|
||
|
||
# Kiểm tra xem user có phải admin không để hiển thị file TXT trên result page
|
||
is_admin = False
|
||
decrypted_files = []
|
||
if session_id in sessions:
|
||
user_info = sessions[session_id].get('user_info')
|
||
if user_info and user_info.get('role') == 'admin':
|
||
is_admin = True
|
||
# Lấy danh sách file TXT đã giải mã cho admin
|
||
for enc_file in encrypted_files:
|
||
if enc_file.filename:
|
||
dec_filename = enc_file.filename.replace('.enc', '')
|
||
txt_path = os.path.join(OUTPUT_FOLDER, dec_filename)
|
||
if os.path.exists(txt_path):
|
||
decrypted_files.append(dec_filename)
|
||
|
||
# Tạo response với cookie
|
||
session_user = sessions[session_id].get('user_info') if session_id in sessions else None
|
||
html_response = templates.TemplateResponse(
|
||
"result.html",
|
||
{
|
||
"request": request,
|
||
"output_files": output_files,
|
||
"decrypted_files": decrypted_files if is_admin else [],
|
||
"is_admin": is_admin,
|
||
"system_info_list": system_info_list,
|
||
"email_sent": email_sent,
|
||
"is_logged_in": session_user is not None,
|
||
"user_info": session_user,
|
||
"session_mode": False,
|
||
"message": {
|
||
"type": "success",
|
||
"text": (
|
||
f"Đã xử lý thành công {len(output_files)} file. ✉️ Đã gửi email thông báo."
|
||
if email_sent else
|
||
f"Đã xử lý thành công {len(output_files)} file. ⚠️ Gửi email thất bại."
|
||
if _email_enabled() else
|
||
f"Đã xử lý thành công {len(output_files)} file."
|
||
)
|
||
}
|
||
}
|
||
)
|
||
|
||
# Set session cookie (expires in 24 hours)
|
||
html_response.set_cookie(
|
||
key="session_id",
|
||
value=session_id,
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600,
|
||
httponly=True,
|
||
samesite='lax'
|
||
)
|
||
|
||
return html_response
|
||
else:
|
||
return templates.TemplateResponse(
|
||
"index.html",
|
||
{
|
||
"request": request,
|
||
"os_config": OS_CONFIG,
|
||
"is_logged_in": _is_logged_in_ctx,
|
||
"user_info": _user_info_ctx,
|
||
"tools_list": get_available_tools(),
|
||
"root_path": get_app_root_path(request),
|
||
"user_email": sessions.get(session_id, {}).get("user_email", ""),
|
||
"message": {"type": "warning", "text": "Processing completed but no output files were generated."}
|
||
}
|
||
)
|
||
|
||
except Exception as e:
|
||
import traceback
|
||
print(Fore.RED + f"Error: {str(e)}" + Fore.RESET)
|
||
print(traceback.format_exc())
|
||
return templates.TemplateResponse(
|
||
"index.html",
|
||
{
|
||
"request": request,
|
||
"os_config": OS_CONFIG,
|
||
"is_logged_in": locals().get('_is_logged_in_ctx', False),
|
||
"user_info": locals().get('_user_info_ctx', None),
|
||
"tools_list": get_available_tools(),
|
||
"root_path": get_app_root_path(request),
|
||
"user_email": sessions.get(session_id, {}).get("user_email", ""),
|
||
"message": {"type": "error", "text": f"Error processing files: {str(e)}"}
|
||
}
|
||
)
|
||
|
||
@app.get("/download/{filename}", name="download_file")
|
||
async def download_file(
|
||
request: Request,
|
||
filename: str
|
||
):
|
||
"""
|
||
Download file - Cho phép download nếu:
|
||
1. File thuộc session của user (vừa upload) - KHÔNG CẦN LOGIN
|
||
2. Hoặc user đã xác thực - CẦN LOGIN
|
||
"""
|
||
try:
|
||
file_path = os.path.join(OUTPUT_FOLDER, filename)
|
||
if not os.path.exists(file_path):
|
||
raise HTTPException(status_code=404, detail="File not found")
|
||
|
||
# Kiểm tra session trước - KHÔNG CẦN AUTHENTICATION
|
||
session_id = request.cookies.get('session_id')
|
||
if session_id and can_access_file(session_id, filename):
|
||
print(Fore.GREEN + f"[DOWNLOAD] ✓ Session access granted for file: {filename} (session: {session_id[:8]}...)" + Fore.RESET)
|
||
return FileResponse(
|
||
file_path,
|
||
media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||
filename=filename
|
||
)
|
||
|
||
# Nếu không có session hoặc file không thuộc session, kiểm tra session login
|
||
print(Fore.YELLOW + f"[DOWNLOAD] File not in session, checking authentication..." + Fore.RESET)
|
||
|
||
# Kiểm tra session login (user đã đăng nhập qua form)
|
||
user = get_session_user(request)
|
||
if user:
|
||
print(Fore.GREEN + f"[DOWNLOAD] ✓ Session login access granted for file: {filename} (user: {user['username']})" + Fore.RESET)
|
||
return FileResponse(
|
||
file_path,
|
||
media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||
filename=filename
|
||
)
|
||
|
||
# Fallback: Kiểm tra HTTP Basic Auth
|
||
auth_header = request.headers.get('Authorization')
|
||
if not auth_header:
|
||
print(Fore.RED + f"[DOWNLOAD] ✗ No session and no authentication for file: {filename}" + Fore.RESET)
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="Authentication required to download this file",
|
||
headers={"WWW-Authenticate": "Basic"},
|
||
)
|
||
|
||
# Parse credentials manually
|
||
try:
|
||
scheme, credentials_str = auth_header.split()
|
||
if scheme.lower() != 'basic':
|
||
raise HTTPException(status_code=401, detail="Invalid authentication scheme")
|
||
|
||
import base64
|
||
decoded = base64.b64decode(credentials_str).decode('utf-8')
|
||
username, password = decoded.split(':', 1)
|
||
|
||
credentials = HTTPBasicCredentials(username=username, password=password)
|
||
user = verify_credentials(credentials)
|
||
|
||
print(Fore.CYAN + f"[DOWNLOAD] ✓ User '{user['username']}' authenticated, downloading: {filename}" + Fore.RESET)
|
||
return FileResponse(
|
||
file_path,
|
||
media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||
filename=filename
|
||
)
|
||
except Exception as e:
|
||
print(Fore.RED + f"[DOWNLOAD] ✗ Authentication failed: {str(e)}" + Fore.RESET)
|
||
raise HTTPException(
|
||
status_code=status.HTTP_401_UNAUTHORIZED,
|
||
detail="Invalid credentials",
|
||
headers={"WWW-Authenticate": "Basic"},
|
||
)
|
||
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.get("/my_files", response_class=HTMLResponse, name="my_files")
|
||
async def my_files(request: Request):
|
||
"""List files của session hiện tại - không cần authentication"""
|
||
try:
|
||
session_id = get_or_create_session(request)
|
||
user = sessions.get(session_id, {}).get('user_info')
|
||
is_logged_in = user is not None
|
||
|
||
# Lấy danh sách files của session (cho guest)
|
||
session_files = sessions.get(session_id, {}).get('files', [])
|
||
|
||
# Nếu đã đăng nhập, lấy thêm files từ Database để quản lý theo User
|
||
if is_logged_in:
|
||
try:
|
||
db = SessionLocal()
|
||
records = db.query(UserFileRecord).filter_by(username=user['username']).order_by(UserFileRecord.created_at.desc()).all()
|
||
db_files = [r.filename for r in records]
|
||
# Merge không trùng lặp
|
||
session_files = list(set(session_files + db_files))
|
||
except Exception as e:
|
||
print(Fore.RED + f"[DB ERROR] Lỗi lấy danh sách file của user: {e}" + Fore.RESET)
|
||
finally:
|
||
db.close()
|
||
|
||
# Kiểm tra files có tồn tại không
|
||
existing_files = []
|
||
for filename in session_files:
|
||
file_path = os.path.join(OUTPUT_FOLDER, filename)
|
||
if os.path.exists(file_path):
|
||
existing_files.append(filename)
|
||
|
||
if existing_files:
|
||
if is_logged_in:
|
||
message_text = f"Bạn có {len(existing_files)} file trong hệ thống."
|
||
else:
|
||
message_text = f"Bạn có {len(existing_files)} file trong session hiện tại."
|
||
else:
|
||
if is_logged_in:
|
||
message_text = "Bạn chưa có file nào trên hệ thống."
|
||
else:
|
||
message_text = "Không tìm thấy file nào trong session của bạn."
|
||
|
||
import asyncio
|
||
all_files_info = await asyncio.to_thread(extract_info_from_output_files)
|
||
user_files_info = [f for f in all_files_info if f.get('filename') in existing_files]
|
||
|
||
response = templates.TemplateResponse(
|
||
"result.html",
|
||
{
|
||
"request": request,
|
||
"output_files": existing_files,
|
||
"files_info": user_files_info,
|
||
"message": {"type": "info" if existing_files else "warning", "text": message_text},
|
||
"session_mode": True,
|
||
"is_logged_in": is_logged_in,
|
||
"user_info": user
|
||
}
|
||
)
|
||
if request.cookies.get('session_id') != session_id:
|
||
response.set_cookie(
|
||
key="session_id",
|
||
value=session_id,
|
||
max_age=SESSION_TIMEOUT_HOURS * 3600,
|
||
httponly=True,
|
||
samesite='lax'
|
||
)
|
||
return response
|
||
except Exception as e:
|
||
return templates.TemplateResponse(
|
||
"index.html",
|
||
{
|
||
"request": request,
|
||
"os_config": OS_CONFIG,
|
||
"message": {"type": "error", "text": f"Error: {str(e)}"}
|
||
}
|
||
)
|
||
|
||
@app.get("/list_outputs", response_class=HTMLResponse, name="list_outputs")
|
||
async def list_outputs(request: Request):
|
||
"""List ALL output files - requires authentication (session-based)"""
|
||
try:
|
||
# Kiểm tra session login trước (không popup)
|
||
user = get_session_user(request)
|
||
if not user:
|
||
# Chưa đăng nhập → redirect về trang login
|
||
return RedirectResponse(url=request.url_for('login'), status_code=303)
|
||
|
||
is_admin = user.get('role') == 'admin'
|
||
import asyncio
|
||
all_files_info = await asyncio.to_thread(extract_info_from_output_files)
|
||
files = [f.get('filename') for f in all_files_info if f.get('filename')]
|
||
txt_files = [f for f in os.listdir(OUTPUT_FOLDER) if f.endswith('.txt')]
|
||
files_info = all_files_info
|
||
|
||
return templates.TemplateResponse(
|
||
"result.html",
|
||
{
|
||
"request": request,
|
||
"output_files": files,
|
||
"decrypted_files": txt_files if is_admin else [],
|
||
"files_info": files_info,
|
||
"show_report_builder": is_admin,
|
||
"is_admin": is_admin,
|
||
"message": {"type": "success", "text": f"Logged in as: {user['username']} ({user['role']})"},
|
||
"user_info": user,
|
||
"session_mode": False,
|
||
"is_logged_in": True,
|
||
"page_name": "admin",
|
||
"admin_tab": "files"
|
||
}
|
||
)
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
return templates.TemplateResponse(
|
||
"index.html",
|
||
{
|
||
"request": request,
|
||
"os_config": OS_CONFIG,
|
||
"message": {"type": "error", "text": f"Error listing files: {str(e)}"}
|
||
}
|
||
)
|
||
# Tools folder path
|
||
TOOLS_FOLDER = 'tools'
|
||
os.makedirs(TOOLS_FOLDER, exist_ok=True)
|
||
|
||
# OS metadata for icons and display names
|
||
OS_METADATA = {
|
||
'ubuntu': {'icon': '🟠', 'display_name': 'Ubuntu'},
|
||
'centos': {'icon': '🔵', 'display_name': 'CentOS'},
|
||
'rhel': {'icon': '🎩', 'display_name': 'RHEL'},
|
||
'oracle': {'icon': '🔸', 'display_name': 'Oracle Linux'},
|
||
'windows': {'icon': '🪟', 'display_name': 'Windows Server'},
|
||
}
|
||
|
||
# Tools download configuration - nested structure: {os_key: {versions: {version_key: tool_data}}}
|
||
TOOLS_CONFIG = {
|
||
'ubuntu': {
|
||
'icon': '🟠',
|
||
'display_name': 'Ubuntu',
|
||
'versions': {
|
||
'ubuntu_2204': {
|
||
'name': 'Ubuntu 22.04 Hardening',
|
||
'file': 'Hardenning_Ubuntu2204_2.0.0.zip',
|
||
'version': 'v2.0.0',
|
||
'updated': '03/02/2026',
|
||
'os_version_label': '22.04',
|
||
'source_path': 'tools/Hardenning_Ubuntu2204_2.0.0.zip'
|
||
}
|
||
}
|
||
},
|
||
'centos': {
|
||
'icon': '🔵',
|
||
'display_name': 'CentOS',
|
||
'versions': {}
|
||
},
|
||
'rhel': {
|
||
'icon': '🎩',
|
||
'display_name': 'RHEL',
|
||
'versions': {}
|
||
},
|
||
'oracle': {
|
||
'icon': '🔸',
|
||
'display_name': 'Oracle Linux',
|
||
'versions': {}
|
||
},
|
||
'windows': {
|
||
'icon': '🪟',
|
||
'display_name': 'Windows Server',
|
||
'versions': {}
|
||
}
|
||
}
|
||
|
||
def save_tools_config():
|
||
"""Save TOOLS_CONFIG to JSON file for persistence"""
|
||
config_file = os.path.join(TOOLS_FOLDER, 'tools_config.json')
|
||
try:
|
||
with open(config_file, 'w', encoding='utf-8') as f:
|
||
json.dump(TOOLS_CONFIG, f, indent=2, ensure_ascii=False)
|
||
print(Fore.GREEN + f"[TOOLS] Saved config to {config_file}" + Fore.RESET)
|
||
except Exception as e:
|
||
print(Fore.RED + f"[TOOLS] Failed to save config: {e}" + Fore.RESET)
|
||
|
||
def load_tools_config():
|
||
"""Load TOOLS_CONFIG from JSON file with backward-compatible migration"""
|
||
global TOOLS_CONFIG
|
||
config_file = os.path.join(TOOLS_FOLDER, 'tools_config.json')
|
||
try:
|
||
if os.path.exists(config_file):
|
||
with open(config_file, 'r', encoding='utf-8') as f:
|
||
loaded_config = json.load(f)
|
||
|
||
# Check if old flat format (has 'file' key directly in os_key level)
|
||
first_key = next(iter(loaded_config), None)
|
||
if first_key and 'file' in loaded_config.get(first_key, {}):
|
||
# Migrate old flat format to new nested format
|
||
print(Fore.YELLOW + "[TOOLS] Migrating old config format to new multi-version format..." + Fore.RESET)
|
||
for os_key, tool_data in loaded_config.items():
|
||
if os_key not in TOOLS_CONFIG:
|
||
meta = OS_METADATA.get(os_key, {'icon': '📦', 'display_name': os_key.upper()})
|
||
TOOLS_CONFIG[os_key] = {
|
||
'icon': meta['icon'],
|
||
'display_name': meta['display_name'],
|
||
'versions': {}
|
||
}
|
||
# Create a version key from old data
|
||
version_key = f"{os_key}_default"
|
||
TOOLS_CONFIG[os_key]['versions'][version_key] = {
|
||
'name': tool_data.get('name', ''),
|
||
'file': tool_data.get('file', ''),
|
||
'version': tool_data.get('version', ''),
|
||
'updated': tool_data.get('updated', ''),
|
||
'os_version_label': '',
|
||
'source_path': tool_data.get('source_path')
|
||
}
|
||
# Save migrated config
|
||
save_tools_config()
|
||
else:
|
||
# New format - merge with defaults
|
||
for os_key, os_data in loaded_config.items():
|
||
if os_key in TOOLS_CONFIG:
|
||
TOOLS_CONFIG[os_key]['icon'] = os_data.get('icon', TOOLS_CONFIG[os_key].get('icon', '📦'))
|
||
TOOLS_CONFIG[os_key]['display_name'] = os_data.get('display_name', TOOLS_CONFIG[os_key].get('display_name', os_key))
|
||
# Merge versions
|
||
for ver_key, ver_data in os_data.get('versions', {}).items():
|
||
TOOLS_CONFIG[os_key]['versions'][ver_key] = ver_data
|
||
else:
|
||
TOOLS_CONFIG[os_key] = os_data
|
||
|
||
print(Fore.GREEN + f"[TOOLS] Loaded config from {config_file}" + Fore.RESET)
|
||
except Exception as e:
|
||
print(Fore.YELLOW + f"[TOOLS] Using default config: {e}" + Fore.RESET)
|
||
|
||
# Load saved tools configuration on startup
|
||
load_tools_config()
|
||
|
||
def get_available_tools():
|
||
"""Get flat list of all tool versions with availability status for templates"""
|
||
tools_list = []
|
||
base_dir = os.path.dirname(os.path.abspath(__file__)) if '__file__' in dir() else '.'
|
||
for os_key, os_data in TOOLS_CONFIG.items():
|
||
icon = os_data.get('icon', '📦')
|
||
display_name = os_data.get('display_name', os_key.upper())
|
||
for ver_key, tool in os_data.get('versions', {}).items():
|
||
source_path = tool.get('source_path')
|
||
available = False
|
||
if source_path:
|
||
full_path = os.path.join(base_dir, source_path)
|
||
available = os.path.exists(full_path)
|
||
|
||
tools_list.append({
|
||
'os_key': os_key,
|
||
'version_key': ver_key,
|
||
'icon': icon,
|
||
'display_name': display_name,
|
||
'os_version_label': tool.get('os_version_label', ''),
|
||
'name': tool['name'],
|
||
'file': tool.get('file', ''),
|
||
'version': tool.get('version', ''),
|
||
'updated': tool.get('updated', ''),
|
||
'available': available
|
||
})
|
||
return tools_list
|
||
|
||
def generate_version_key(os_type: str, os_version_label: str) -> str:
|
||
"""Generate a unique version key from OS type and version label"""
|
||
# Sanitize: replace dots and spaces with underscores
|
||
sanitized = os_version_label.replace('.', '').replace(' ', '_').lower()
|
||
return f"{os_type}_{sanitized}"
|
||
|
||
@app.get("/download/tools/{version_key}", name="download_tool")
|
||
async def download_tool(version_key: str):
|
||
"""Download audit tool script for specific OS version"""
|
||
# Find the tool by version_key
|
||
tool = None
|
||
for os_key, os_data in TOOLS_CONFIG.items():
|
||
if version_key in os_data.get('versions', {}):
|
||
tool = os_data['versions'][version_key]
|
||
break
|
||
|
||
if not tool:
|
||
raise HTTPException(status_code=404, detail=f"Tool '{version_key}' not found")
|
||
|
||
source_path = tool.get('source_path')
|
||
if not source_path:
|
||
raise HTTPException(status_code=404, detail=f"Tool '{version_key}' is not available yet")
|
||
|
||
# Resolve full path
|
||
full_path = os.path.join(os.path.dirname(__file__), source_path)
|
||
full_path = os.path.normpath(full_path)
|
||
|
||
if not os.path.exists(full_path):
|
||
raise HTTPException(status_code=404, detail=f"Tool file not found: {tool.get('file', '')}")
|
||
|
||
print(Fore.GREEN + f"[DOWNLOAD] User downloading tool: {tool['name']} ({tool.get('version', '')})" + Fore.RESET)
|
||
|
||
return FileResponse(
|
||
path=full_path,
|
||
filename=tool.get('file', os.path.basename(full_path)),
|
||
media_type='application/octet-stream'
|
||
)
|
||
|
||
@app.get("/api/tools")
|
||
async def get_tools_info():
|
||
"""Get information about available audit tools"""
|
||
return get_available_tools()
|
||
|
||
# ────────────────────────────────────────────────────────────────────────────
|
||
# Documentation Guidelines Configuration Management
|
||
# ────────────────────────────────────────────────────────────────────────────
|
||
DOCS_FOLDER = 'docs'
|
||
os.makedirs(DOCS_FOLDER, exist_ok=True)
|
||
|
||
DOCS_CONFIG = {}
|
||
|
||
def save_docs_config():
|
||
"""Save DOCS_CONFIG to JSON file for persistence"""
|
||
config_file = os.path.join(DOCS_FOLDER, 'docs_config.json')
|
||
try:
|
||
with open(config_file, 'w', encoding='utf-8') as f:
|
||
json.dump(DOCS_CONFIG, f, indent=2, ensure_ascii=False)
|
||
print(Fore.GREEN + f"[DOCS] Saved config to {config_file}" + Fore.RESET)
|
||
except Exception as e:
|
||
print(Fore.RED + f"[DOCS] Failed to save config: {e}" + Fore.RESET)
|
||
|
||
def load_docs_config():
|
||
"""Load DOCS_CONFIG from JSON file"""
|
||
global DOCS_CONFIG
|
||
config_file = os.path.join(DOCS_FOLDER, 'docs_config.json')
|
||
try:
|
||
if os.path.exists(config_file):
|
||
with open(config_file, 'r', encoding='utf-8') as f:
|
||
DOCS_CONFIG = json.load(f)
|
||
print(Fore.GREEN + f"[DOCS] Loaded config from {config_file}" + Fore.RESET)
|
||
except Exception as e:
|
||
print(Fore.YELLOW + f"[DOCS] Failed to load config: {e}" + Fore.RESET)
|
||
|
||
load_docs_config()
|
||
|
||
def get_available_docs():
|
||
"""Get list of documentation files with availability status for templates"""
|
||
docs_list = []
|
||
base_dir = os.path.dirname(os.path.abspath(__file__)) if '__file__' in dir() else '.'
|
||
for doc_key, doc in DOCS_CONFIG.items():
|
||
source_path = doc.get('source_path')
|
||
available = False
|
||
if source_path:
|
||
full_path = os.path.join(base_dir, source_path)
|
||
available = os.path.exists(full_path)
|
||
|
||
docs_list.append({
|
||
'doc_key': doc_key,
|
||
'name': doc.get('name', ''),
|
||
'file': doc.get('file', ''),
|
||
'version': doc.get('version', ''),
|
||
'updated': doc.get('updated', ''),
|
||
'size': doc.get('size', ''),
|
||
'available': available
|
||
})
|
||
return docs_list
|
||
|
||
@app.get("/download/docs/{doc_key}", name="download_doc")
|
||
async def download_doc(doc_key: str):
|
||
"""Download configuration documentation file"""
|
||
doc = DOCS_CONFIG.get(doc_key)
|
||
if not doc:
|
||
raise HTTPException(status_code=404, detail=f"Document '{doc_key}' not found")
|
||
|
||
source_path = doc.get('source_path')
|
||
if not source_path:
|
||
raise HTTPException(status_code=404, detail="Document file path not set")
|
||
|
||
full_path = os.path.join(os.path.dirname(__file__), source_path)
|
||
full_path = os.path.normpath(full_path)
|
||
|
||
if not os.path.exists(full_path):
|
||
raise HTTPException(status_code=404, detail="Document file not found on server")
|
||
|
||
print(Fore.GREEN + f"[DOWNLOAD] User downloading doc: {doc['name']} ({doc.get('version', '')})" + Fore.RESET)
|
||
return FileResponse(
|
||
path=full_path,
|
||
filename=doc.get('file', os.path.basename(full_path)),
|
||
media_type='application/octet-stream'
|
||
)
|
||
|
||
# ============== ADMIN ROUTES ==============
|
||
|
||
def require_admin_session(request: Request):
|
||
"""Check session auth and return admin user or None"""
|
||
user = get_session_user(request)
|
||
if user and user.get('role') == 'admin':
|
||
return user
|
||
return None
|
||
|
||
@app.get("/admin/tools", response_class=HTMLResponse, name="admin_tools")
|
||
async def admin_tools_page(request: Request, msg: str = None, msg_type: str = "success"):
|
||
"""Admin page for managing tools - requires session authentication"""
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
return RedirectResponse(url=request.url_for('login'), status_code=303)
|
||
|
||
# Read flash message from query params (PRG pattern)
|
||
message = None
|
||
if msg:
|
||
message = {"type": msg_type, "text": msg}
|
||
|
||
# Lấy thông tin thống kê email
|
||
db = SessionLocal()
|
||
email_stats = []
|
||
try:
|
||
records = db.query(UserEmailRecord).order_by(UserEmailRecord.last_seen.desc()).all()
|
||
for r in records:
|
||
email_stats.append({
|
||
"email": r.email,
|
||
"display_name": r.display_name,
|
||
"total_uploads": r.total_uploads,
|
||
"total_files": r.total_files,
|
||
"first_seen": r.first_seen.strftime("%d/%m/%Y %H:%M") if r.first_seen else "",
|
||
"last_seen": r.last_seen.strftime("%d/%m/%Y %H:%M") if r.last_seen else ""
|
||
})
|
||
finally:
|
||
db.close()
|
||
|
||
# Load unit mapping data for the admin UI
|
||
unit_mapping_data = _load_unit_mapping()
|
||
|
||
return templates.TemplateResponse(
|
||
"admin_tools.html",
|
||
{
|
||
"request": request,
|
||
"tools_list": get_available_tools(),
|
||
"os_metadata": OS_METADATA,
|
||
"message": message,
|
||
"user": user,
|
||
"user_info": user,
|
||
"is_logged_in": True,
|
||
"email_module_available": EMAIL_MODULE_AVAILABLE,
|
||
"email_enabled": email_enabled_runtime,
|
||
"email_stats": email_stats,
|
||
"unit_mapping_units": unit_mapping_data.get("units", {}),
|
||
"fixed_unit_override": unit_mapping_data.get("fixed_unit_override", {}),
|
||
"unit_mapping_group_mapping": unit_mapping_data.get("group_mapping", {}),
|
||
"unit_mapping_users": unit_mapping_data.get("users", {}),
|
||
"page_name": "admin",
|
||
"admin_tab": "tools"
|
||
}
|
||
)
|
||
|
||
@app.get("/admin/users", response_class=HTMLResponse, name="admin_users_page")
|
||
async def admin_users_page(request: Request, msg: str = None, msg_type: str = "success"):
|
||
"""Admin page for managing users and unit mapping"""
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
return RedirectResponse(url=request.url_for('login'), status_code=303)
|
||
|
||
message = None
|
||
if msg:
|
||
message = {"type": msg_type, "text": msg}
|
||
|
||
unit_mapping_data = _load_unit_mapping()
|
||
|
||
return templates.TemplateResponse(
|
||
"admin_users.html",
|
||
{
|
||
"request": request,
|
||
"message": message,
|
||
"user": user,
|
||
"user_info": user,
|
||
"is_logged_in": True,
|
||
"unit_mapping_units": unit_mapping_data.get("units", {}),
|
||
"fixed_unit_override": unit_mapping_data.get("fixed_unit_override", {}),
|
||
"unit_mapping_group_mapping": unit_mapping_data.get("group_mapping", {}),
|
||
"unit_mapping_users": unit_mapping_data.get("users", {}),
|
||
"page_name": "admin",
|
||
"admin_tab": "users"
|
||
}
|
||
)
|
||
|
||
@app.post("/admin/email/enable", name="admin_email_enable")
|
||
async def admin_email_enable(request: Request):
|
||
"""Bật chức năng gửi email (chỉ kịp thời, reset khi restart server)"""
|
||
global email_enabled_runtime
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
return RedirectResponse(url=request.url_for('login'), status_code=303)
|
||
if not EMAIL_MODULE_AVAILABLE:
|
||
params = "msg=Module+email+kh%C3%B4ng+kh%E1%BA%A3+d%E1%BB%A5ng+tr%C3%AAn+server+n%C3%A0y&msg_type=error"
|
||
return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303)
|
||
email_enabled_runtime = True
|
||
print(Fore.GREEN + f"[EMAIL] Admin '{user.get('username')}' ENABLED email notifications" + Fore.RESET)
|
||
params = "msg=Đã+bật+gửi+email+thông+báo&msg_type=success"
|
||
return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303)
|
||
|
||
@app.post("/admin/email/disable", name="admin_email_disable")
|
||
async def admin_email_disable(request: Request):
|
||
"""Tắt chức năng gửi email"""
|
||
global email_enabled_runtime
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
return RedirectResponse(url=request.url_for('login'), status_code=303)
|
||
email_enabled_runtime = False
|
||
print(Fore.YELLOW + f"[EMAIL] Admin '{user.get('username')}' DISABLED email notifications" + Fore.RESET)
|
||
params = "msg=Đã+tắt+gửi+email+thông+báo&msg_type=success"
|
||
return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303)
|
||
|
||
@app.get("/admin/email-stats", name="admin_email_stats")
|
||
async def admin_email_stats(request: Request):
|
||
"""Trả về JSON danh sách email + thống kê upload (chỉ admin)."""
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
raise HTTPException(status_code=403, detail="Admin only")
|
||
db = SessionLocal()
|
||
try:
|
||
records = db.query(UserEmailRecord).order_by(UserEmailRecord.last_seen.desc()).all()
|
||
return [
|
||
{
|
||
"email": r.email,
|
||
"display_name": r.display_name,
|
||
"total_uploads": r.total_uploads,
|
||
"total_files": r.total_files,
|
||
"first_seen": r.first_seen.strftime("%d/%m/%Y %H:%M") if r.first_seen else "",
|
||
"last_seen": r.last_seen.strftime("%d/%m/%Y %H:%M") if r.last_seen else "",
|
||
}
|
||
for r in records
|
||
]
|
||
finally:
|
||
db.close()
|
||
|
||
@app.delete("/api/files/{filename}")
|
||
async def delete_file(filename: str, request: Request):
|
||
"""Xóa file. Chỉ admin mới có quyền thực hiện."""
|
||
user = get_session_user(request)
|
||
if not user or user.get('role') != 'admin':
|
||
raise HTTPException(status_code=403, detail="Chỉ admin mới có quyền xóa file.")
|
||
|
||
deleted = False
|
||
|
||
# Xóa file kết quả
|
||
dec_path = os.path.join(OUTPUT_FOLDER, filename)
|
||
if os.path.exists(dec_path):
|
||
os.remove(dec_path)
|
||
deleted = True
|
||
|
||
# Xóa file mã hóa (thường có thêm đuôi .enc)
|
||
enc_path = os.path.join(UPLOAD_FOLDER, filename + '.enc')
|
||
if os.path.exists(enc_path):
|
||
os.remove(enc_path)
|
||
deleted = True
|
||
|
||
# Xóa trong database
|
||
try:
|
||
db = SessionLocal()
|
||
|
||
# Xóa khỏi UserFileRecord
|
||
records = db.query(UserFileRecord).filter_by(filename=filename).all()
|
||
for r in records:
|
||
db.delete(r)
|
||
deleted = True
|
||
|
||
# Xóa khỏi AuditRecord
|
||
audit_recs = db.query(models.AuditRecord).filter_by(filename=filename).all()
|
||
for ar in audit_recs:
|
||
if ar.has_txt and ar.txt_filename:
|
||
for p in [os.path.join(OUTPUT_FOLDER, ar.txt_filename), os.path.join(UPLOAD_FOLDER, ar.txt_filename + '.enc')]:
|
||
if os.path.exists(p):
|
||
try: os.remove(p)
|
||
except Exception: pass
|
||
db.delete(ar)
|
||
deleted = True
|
||
|
||
db.commit()
|
||
except Exception as e:
|
||
print(Fore.RED + f"Error deleting file record: {e}" + Fore.RESET)
|
||
finally:
|
||
db.close()
|
||
|
||
if deleted:
|
||
return {"message": f"Đã xóa file {filename} thành công"}
|
||
else:
|
||
# File có thể đã bị xóa trước đó
|
||
return {"message": "File không tồn tại hoặc đã bị xóa"}
|
||
|
||
@app.get("/api/admin/units")
|
||
async def api_admin_units(request: Request):
|
||
"""API: Lấy danh sách đơn vị từ unit_mapping.json. Chỉ admin."""
|
||
user = get_session_user(request)
|
||
if not user or user.get('role') != 'admin':
|
||
raise HTTPException(status_code=403, detail="Chỉ admin mới có quyền truy cập.")
|
||
|
||
mapping = _load_unit_mapping()
|
||
units = mapping.get('units', {})
|
||
result = [{"unit_id": str(uid), "unit_name": uname} for uid, uname in units.items()]
|
||
return result
|
||
|
||
|
||
@app.put("/api/admin/files/{filename}/unit")
|
||
async def api_admin_update_file_unit(filename: str, request: Request):
|
||
"""API: Cập nhật đơn vị của một file. Chỉ admin."""
|
||
user = get_session_user(request)
|
||
if not user or user.get('role') != 'admin':
|
||
raise HTTPException(status_code=403, detail="Chỉ admin mới có quyền thực hiện.")
|
||
|
||
try:
|
||
body = await request.json()
|
||
unit_id = str(body.get('unit_id', '')).strip()
|
||
unit_name = body.get('unit_name', '').strip()
|
||
except Exception:
|
||
raise HTTPException(status_code=400, detail="Body không hợp lệ. Cần {unit_id, unit_name}")
|
||
|
||
if not unit_id:
|
||
raise HTTPException(status_code=400, detail="unit_id không được để trống")
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
records = db.query(UserFileRecord).filter_by(filename=filename).all()
|
||
if not records:
|
||
# Tạo mới nếu chưa có
|
||
record = UserFileRecord(
|
||
filename=filename,
|
||
username=user.get('username', 'admin'),
|
||
unit_id=unit_id,
|
||
unit_name=unit_name,
|
||
)
|
||
db.add(record)
|
||
db.commit()
|
||
return {"message": f"Đã tạo bản ghi và gán đơn vị '{unit_name}' cho file '{filename}'"}
|
||
|
||
for record in records:
|
||
record.unit_id = unit_id
|
||
record.unit_name = unit_name
|
||
db.commit()
|
||
return {"message": f"Đã cập nhật đơn vị '{unit_name}' cho {len(records)} bản ghi của file '{filename}'"}
|
||
except Exception as e:
|
||
db.rollback()
|
||
raise HTTPException(status_code=500, detail=f"Lỗi DB: {str(e)}")
|
||
finally:
|
||
db.close()
|
||
|
||
@app.post("/admin/tools/upload", name="admin_tools_upload")
|
||
async def admin_upload_tool(
|
||
request: Request,
|
||
os_type: str = Form(...),
|
||
os_version_label: str = Form(...),
|
||
version: str = Form(...),
|
||
tool_name: str = Form(...),
|
||
tool_file: UploadFile = File(...)
|
||
):
|
||
"""Upload a new tool - requires admin session authentication"""
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
return RedirectResponse(url=request.url_for('login'), status_code=303)
|
||
|
||
try:
|
||
# Save file to tools folder
|
||
filename = secure_filename(tool_file.filename)
|
||
file_path = os.path.join(TOOLS_FOLDER, filename)
|
||
|
||
with open(file_path, "wb") as buffer:
|
||
content = await tool_file.read()
|
||
buffer.write(content)
|
||
|
||
# Generate version key
|
||
version_key = generate_version_key(os_type, os_version_label)
|
||
|
||
# Ensure OS key exists in TOOLS_CONFIG
|
||
if os_type not in TOOLS_CONFIG:
|
||
meta = OS_METADATA.get(os_type, {'icon': '📦', 'display_name': os_type.upper()})
|
||
TOOLS_CONFIG[os_type] = {
|
||
'icon': meta['icon'],
|
||
'display_name': meta['display_name'],
|
||
'versions': {}
|
||
}
|
||
|
||
# Add/update version entry
|
||
today = datetime.now().strftime("%d/%m/%Y")
|
||
TOOLS_CONFIG[os_type]['versions'][version_key] = {
|
||
'name': tool_name,
|
||
'file': filename,
|
||
'version': version,
|
||
'updated': today,
|
||
'os_version_label': os_version_label,
|
||
'source_path': f'tools/{filename}'
|
||
}
|
||
|
||
# Save config
|
||
save_tools_config()
|
||
|
||
print(Fore.GREEN + f"[ADMIN] Tool uploaded: {filename} for {os_type} {os_version_label}" + Fore.RESET)
|
||
|
||
# PRG: Redirect to avoid form resubmission and path issues
|
||
from urllib.parse import urlencode
|
||
params = urlencode({"msg": f"Đã tải lên công cụ '{tool_name}' ({os_version_label}) thành công!", "msg_type": "success"})
|
||
return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303)
|
||
except Exception as e:
|
||
print(Fore.RED + f"[ADMIN] Upload error: {e}" + Fore.RESET)
|
||
from urllib.parse import urlencode
|
||
params = urlencode({"msg": f"Lỗi khi tải lên: {str(e)}", "msg_type": "error"})
|
||
return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303)
|
||
|
||
@app.post("/admin/tools/delete/{version_key}", name="admin_tools_delete")
|
||
async def admin_delete_tool(
|
||
request: Request,
|
||
version_key: str
|
||
):
|
||
"""Delete a tool version - requires admin session authentication"""
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
return RedirectResponse(url=request.url_for('login'), status_code=303)
|
||
|
||
try:
|
||
# Find the version in TOOLS_CONFIG
|
||
found_os_key = None
|
||
for os_key, os_data in TOOLS_CONFIG.items():
|
||
if version_key in os_data.get('versions', {}):
|
||
found_os_key = os_key
|
||
break
|
||
|
||
if not found_os_key:
|
||
raise HTTPException(status_code=404, detail="Tool version not found")
|
||
|
||
tool = TOOLS_CONFIG[found_os_key]['versions'][version_key]
|
||
source_path = tool.get('source_path')
|
||
tool_name = tool.get('name', version_key)
|
||
|
||
# Delete file if exists
|
||
if source_path:
|
||
file_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), source_path)
|
||
if os.path.exists(file_path):
|
||
os.remove(file_path)
|
||
print(Fore.YELLOW + f"[ADMIN] Deleted file: {file_path}" + Fore.RESET)
|
||
|
||
# Remove version entry from config
|
||
del TOOLS_CONFIG[found_os_key]['versions'][version_key]
|
||
save_tools_config()
|
||
|
||
# PRG: Redirect to avoid stale state
|
||
from urllib.parse import urlencode
|
||
params = urlencode({"msg": f"Đã xóa công cụ '{tool_name}' thành công!", "msg_type": "success"})
|
||
return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303)
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
print(Fore.RED + f"[ADMIN] Delete error: {e}" + Fore.RESET)
|
||
from urllib.parse import urlencode
|
||
params = urlencode({"msg": f"Lỗi khi xóa: {str(e)}", "msg_type": "error"})
|
||
return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303)
|
||
|
||
@app.post("/admin/docs/upload", name="admin_docs_upload")
|
||
async def admin_upload_doc(
|
||
request: Request,
|
||
doc_name: str = Form(...),
|
||
doc_version: str = Form("v1.0"),
|
||
doc_file: UploadFile = File(...)
|
||
):
|
||
"""Upload a new document - requires admin session"""
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
return RedirectResponse(url=request.url_for('login'), status_code=303)
|
||
|
||
try:
|
||
filename = secure_filename(doc_file.filename)
|
||
file_path = os.path.join(DOCS_FOLDER, filename)
|
||
|
||
with open(file_path, "wb") as buffer:
|
||
content = await doc_file.read()
|
||
buffer.write(content)
|
||
|
||
size_bytes = len(content)
|
||
if size_bytes >= 1024 * 1024:
|
||
size_str = f"{size_bytes / (1024*1024):.1f} MB"
|
||
elif size_bytes >= 1024:
|
||
size_str = f"{size_bytes / 1024:.1f} KB"
|
||
else:
|
||
size_str = f"{size_bytes} B"
|
||
|
||
doc_key = f"doc_{int(time.time())}"
|
||
today = datetime.now().strftime("%d/%m/%Y")
|
||
DOCS_CONFIG[doc_key] = {
|
||
'name': doc_name,
|
||
'file': filename,
|
||
'version': doc_version,
|
||
'updated': today,
|
||
'size': size_str,
|
||
'source_path': f'docs/{filename}'
|
||
}
|
||
save_docs_config()
|
||
|
||
print(Fore.GREEN + f"[ADMIN] Doc uploaded: {filename}" + Fore.RESET)
|
||
from urllib.parse import urlencode
|
||
params = urlencode({"msg": f"Đã tải lên tài liệu '{doc_name}' thành công!", "msg_type": "success"})
|
||
return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303)
|
||
except Exception as e:
|
||
print(Fore.RED + f"[ADMIN] Doc upload error: {e}" + Fore.RESET)
|
||
from urllib.parse import urlencode
|
||
params = urlencode({"msg": f"Lỗi khi tải lên tài liệu: {str(e)}", "msg_type": "error"})
|
||
return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303)
|
||
|
||
@app.post("/admin/docs/delete/{doc_key}", name="admin_docs_delete")
|
||
async def admin_delete_doc(request: Request, doc_key: str):
|
||
"""Delete a document - requires admin session"""
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
return RedirectResponse(url=request.url_for('login'), status_code=303)
|
||
|
||
doc = DOCS_CONFIG.get(doc_key)
|
||
if doc:
|
||
filename = doc.get('file', '')
|
||
base_dir = os.path.dirname(os.path.abspath(__file__)) if '__file__' in dir() else '.'
|
||
file_path = os.path.join(base_dir, doc.get('source_path', f'docs/{filename}'))
|
||
if os.path.exists(file_path):
|
||
try:
|
||
os.remove(file_path)
|
||
except Exception:
|
||
pass
|
||
del DOCS_CONFIG[doc_key]
|
||
save_docs_config()
|
||
msg = f"Đã xóa tài liệu '{doc.get('name')}' thành công!"
|
||
msg_type = "success"
|
||
else:
|
||
msg = "Tài liệu không tồn tại hoặc đã bị xóa!"
|
||
msg_type = "error"
|
||
|
||
from urllib.parse import urlencode
|
||
params = urlencode({"msg": msg, "msg_type": msg_type})
|
||
return RedirectResponse(url=f"{request.url_for('admin_tools')}?{params}", status_code=303)
|
||
|
||
# ============== ADMIN UNIT MAPPING API ==============
|
||
|
||
@app.get("/api/admin/unit-mapping", name="api_admin_unit_mapping")
|
||
async def api_get_unit_mapping(request: Request):
|
||
"""Trả về toàn bộ nội dung unit_mapping.json cho admin"""
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
raise HTTPException(status_code=403, detail="Admin required")
|
||
|
||
mapping = _load_unit_mapping()
|
||
return {
|
||
"units": mapping.get("units", {}),
|
||
"fixed_unit_override": mapping.get("fixed_unit_override", {}),
|
||
"group_mapping": mapping.get("group_mapping", {}),
|
||
"users": mapping.get("users", {}),
|
||
"admins": mapping.get("admins", [])
|
||
}
|
||
|
||
@app.post("/api/admin/unit-mapping/fixed-override", name="api_admin_fixed_override_save")
|
||
async def api_save_fixed_override(request: Request):
|
||
"""Thêm/sửa một entry trong fixed_unit_override"""
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
raise HTTPException(status_code=403, detail="Admin required")
|
||
|
||
body = await request.json()
|
||
target_username = body.get("username", "").strip().lower()
|
||
unit_id = body.get("unit_id")
|
||
|
||
if not target_username:
|
||
raise HTTPException(status_code=400, detail="Username không được để trống")
|
||
if unit_id is None or str(unit_id) == "":
|
||
raise HTTPException(status_code=400, detail="Vui lòng chọn đơn vị")
|
||
|
||
try:
|
||
unit_id = int(unit_id)
|
||
except (ValueError, TypeError):
|
||
raise HTTPException(status_code=400, detail="unit_id phải là số nguyên")
|
||
|
||
# Kiểm tra unit_id hợp lệ
|
||
mapping = _load_unit_mapping()
|
||
units = mapping.get("units", {})
|
||
if str(unit_id) not in units:
|
||
raise HTTPException(status_code=400, detail=f"unit_id '{unit_id}' không tồn tại trong danh sách đơn vị")
|
||
|
||
# Đọc file gốc, cập nhật và ghi lại
|
||
try:
|
||
with open(_UNIT_MAPPING_PATH, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
|
||
if "fixed_unit_override" not in data:
|
||
data["fixed_unit_override"] = {}
|
||
|
||
data["fixed_unit_override"][target_username] = unit_id
|
||
|
||
with open(_UNIT_MAPPING_PATH, 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=4)
|
||
|
||
# Reset cache mtime để force reload
|
||
global _unit_mapping_mtime
|
||
_unit_mapping_mtime = 0.0
|
||
|
||
unit_name = units.get(str(unit_id), "")
|
||
print(Fore.GREEN + f"[ADMIN] Fixed unit override: '{target_username}' → '{unit_name}' (ID: {unit_id})" + Fore.RESET)
|
||
|
||
return {"success": True, "message": f"Đã gán '{target_username}' vào đơn vị '{unit_name}'"}
|
||
except Exception as e:
|
||
print(Fore.RED + f"[ADMIN] Error saving fixed override: {e}" + Fore.RESET)
|
||
raise HTTPException(status_code=500, detail=f"Lỗi khi lưu: {str(e)}")
|
||
|
||
@app.delete("/api/admin/unit-mapping/fixed-override/{target_username}", name="api_admin_fixed_override_delete")
|
||
async def api_delete_fixed_override(request: Request, target_username: str):
|
||
"""Xóa một entry trong fixed_unit_override"""
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
raise HTTPException(status_code=403, detail="Admin required")
|
||
|
||
try:
|
||
with open(_UNIT_MAPPING_PATH, 'r', encoding='utf-8') as f:
|
||
data = json.load(f)
|
||
|
||
overrides = data.get("fixed_unit_override", {})
|
||
if target_username not in overrides:
|
||
raise HTTPException(status_code=404, detail=f"User '{target_username}' không có trong danh sách override")
|
||
|
||
del data["fixed_unit_override"][target_username]
|
||
|
||
with open(_UNIT_MAPPING_PATH, 'w', encoding='utf-8') as f:
|
||
json.dump(data, f, ensure_ascii=False, indent=4)
|
||
|
||
# Reset cache mtime để force reload
|
||
global _unit_mapping_mtime
|
||
_unit_mapping_mtime = 0.0
|
||
|
||
print(Fore.YELLOW + f"[ADMIN] Removed fixed unit override for '{target_username}'" + Fore.RESET)
|
||
|
||
return {"success": True, "message": f"Đã xóa override cho '{target_username}'"}
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
print(Fore.RED + f"[ADMIN] Error deleting fixed override: {e}" + Fore.RESET)
|
||
raise HTTPException(status_code=500, detail=f"Lỗi khi xóa: {str(e)}")
|
||
|
||
@app.post("/api/admin/unit-mapping/remap-existing", name="api_admin_remap_existing")
|
||
async def api_remap_existing_users(request: Request):
|
||
"""Quét lại toàn bộ UserFileRecord và tự động map đơn vị dựa trên cấu hình hiện tại"""
|
||
user = require_admin_session(request)
|
||
if not user:
|
||
raise HTTPException(status_code=403, detail="Admin required")
|
||
|
||
mapping = _load_unit_mapping()
|
||
units = mapping.get("units", {})
|
||
fixed_override = mapping.get("fixed_unit_override", {})
|
||
users_fallback = mapping.get("users", {})
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
records = db.query(models.UserFileRecord).all()
|
||
updated_count = 0
|
||
|
||
for record in records:
|
||
if not record.username:
|
||
continue
|
||
|
||
uname = record.username.lower()
|
||
new_unit_id = None
|
||
|
||
# 1. Check fixed override
|
||
if uname in fixed_override:
|
||
new_unit_id = str(fixed_override[uname])
|
||
# 2. Check fallback users
|
||
elif uname in users_fallback:
|
||
new_unit_id = str(users_fallback[uname])
|
||
|
||
if new_unit_id and new_unit_id in units:
|
||
new_unit_name = units[new_unit_id]
|
||
# Chỉ update nếu khác thông tin hiện tại hoặc hiện tại đang trống
|
||
if record.unit_id != new_unit_id or record.unit_name != new_unit_name:
|
||
record.unit_id = new_unit_id
|
||
record.unit_name = new_unit_name
|
||
updated_count += 1
|
||
|
||
if updated_count > 0:
|
||
db.commit()
|
||
|
||
return {"success": True, "message": f"Đã tự động cập nhật lại đơn vị cho {updated_count} file(s)."}
|
||
except Exception as e:
|
||
db.rollback()
|
||
print(Fore.RED + f"[ADMIN] Error remapping users: {e}" + Fore.RESET)
|
||
raise HTTPException(status_code=500, detail=f"Lỗi khi remap: {str(e)}")
|
||
finally:
|
||
db.close()
|
||
|
||
# ============== ADMIN REPORTS ==============
|
||
|
||
def extract_info_from_output_files():
|
||
"""Scan output folder, sync with DB if needed, and return compliance stats from DB."""
|
||
db = SessionLocal()
|
||
try:
|
||
results = []
|
||
base_dir = os.path.abspath(OUTPUT_FOLDER)
|
||
|
||
# List all report files present on disk
|
||
valid_exts = ('.xlsx', '.html', '.docx', '.txt', '.enc_result', '.json', '.csv')
|
||
excel_files = [f for f in os.listdir(base_dir) if f.endswith(valid_exts) and not f.startswith('Tong_hop_') and not f.startswith('Thong_ke_san_luong_') and not f.startswith('~')]
|
||
|
||
# Get existing filenames from DB
|
||
existing_records = db.query(models.AuditRecord.filename).all()
|
||
existing_filenames = {r[0] for r in existing_records}
|
||
|
||
new_files = [f for f in excel_files if f not in existing_filenames]
|
||
|
||
# Process and sync new files (if any were copied manually or created before DB existed)
|
||
for excel_file in new_files:
|
||
excel_path = os.path.join(base_dir, excel_file)
|
||
try:
|
||
file_stat = os.stat(excel_path)
|
||
except Exception:
|
||
continue
|
||
|
||
info = {
|
||
'filename': excel_file,
|
||
'file_size': file_stat.st_size,
|
||
'file_date': datetime.fromtimestamp(file_stat.st_mtime).strftime('%d/%m/%Y %H:%M'),
|
||
'hostname': '',
|
||
'ip_address': '',
|
||
'os_name': '',
|
||
'compliance_percentage': 0.0,
|
||
'passed_count': 0,
|
||
'failed_count': 0,
|
||
'total_checks': 0,
|
||
'mandatory_passed': 0,
|
||
'mandatory_failed': 0,
|
||
'mandatory_total': 0,
|
||
'mandatory_percentage': 0.0,
|
||
'optional_passed': 0,
|
||
'optional_failed': 0,
|
||
'optional_total': 0,
|
||
'has_txt': False,
|
||
'txt_filename': ''
|
||
}
|
||
|
||
# Extract hostname from filename
|
||
base_name = re.sub(r'\.(xlsx|html|docx|txt|enc_result|json|csv)$', '', excel_file, flags=re.IGNORECASE)
|
||
date_match = re.search(r'_(\d{4}_\d{2}_\d{2}-\d{2}_\d{2}_\d{2})$', base_name)
|
||
if date_match:
|
||
prefix = base_name[:date_match.start()]
|
||
# If prefix has an underscore, check if first part is a known OS
|
||
parts = prefix.split('_', 1)
|
||
known_os = ['windows', 'linux', 'ubuntu', 'centos', 'redhat', 'aix', 'solaris', 'hp-ux']
|
||
if len(parts) == 2 and parts[0].lower() in known_os:
|
||
info['os_name'] = parts[0]
|
||
info['hostname'] = parts[1]
|
||
else:
|
||
info['hostname'] = prefix
|
||
else:
|
||
info['hostname'] = base_name
|
||
|
||
# Try to find corresponding TXT file and extract system info
|
||
txt_files = [f for f in os.listdir(base_dir) if f.endswith('.txt')]
|
||
hostname_lower = info['hostname'].lower().replace('-', '').replace('_', '')
|
||
|
||
for txt_file in txt_files:
|
||
txt_lower = txt_file.lower().replace('-', '').replace('_', '')
|
||
if hostname_lower and hostname_lower in txt_lower:
|
||
info['has_txt'] = True
|
||
info['txt_filename'] = txt_file
|
||
# Parse TXT for system info
|
||
try:
|
||
txt_path = os.path.join(base_dir, txt_file)
|
||
with open(txt_path, 'r', encoding='utf-8', errors='ignore') as f:
|
||
lines = f.read().splitlines()
|
||
sys_info = audit_decryption.extract_system_info(lines)
|
||
if sys_info.get('hostname'):
|
||
info['hostname'] = sys_info['hostname']
|
||
if sys_info.get('ip_address'):
|
||
info['ip_address'] = sys_info['ip_address']
|
||
if sys_info.get('os_name'):
|
||
info['os_name'] = sys_info['os_name']
|
||
info['compliance_percentage'] = sys_info.get('compliance_percentage', 0.0)
|
||
info['passed_count'] = sys_info.get('passed_count', 0)
|
||
info['failed_count'] = sys_info.get('failed_count', 0)
|
||
info['total_checks'] = sys_info.get('total_checks', 0)
|
||
except Exception as e:
|
||
print(Fore.YELLOW + f"[REPORTS] Error parsing TXT {txt_file}: {e}" + Fore.RESET)
|
||
break
|
||
|
||
# Parse Excel to get mandatory/optional stats (only for .xlsx files)
|
||
if excel_file.lower().endswith('.xlsx'):
|
||
try:
|
||
from openpyxl import load_workbook
|
||
wb = load_workbook(excel_path, read_only=True, data_only=True)
|
||
ws = wb.worksheets[0]
|
||
|
||
passed = 0
|
||
failed = 0
|
||
mandatory_passed = 0
|
||
mandatory_failed = 0
|
||
optional_passed = 0
|
||
optional_failed = 0
|
||
|
||
for row in ws.iter_rows(min_row=1, max_row=ws.max_row):
|
||
col_c = str(row[2].value).strip().lower() if len(row) > 2 and row[2].value else ''
|
||
col_d = str(row[3].value).strip().lower() if len(row) > 3 and row[3].value else ''
|
||
col_e = str(row[4].value).strip().lower() if len(row) > 4 and row[4].value else ''
|
||
is_mandatory = col_e in ('x', 'v', '1', 'có', 'true', 'yes', 'bắt buộc')
|
||
|
||
if col_c in ('x', 'v', '1', 'passed', 'đạt', 'true', 'ok'):
|
||
passed += 1
|
||
if is_mandatory:
|
||
mandatory_passed += 1
|
||
else:
|
||
optional_passed += 1
|
||
elif col_d in ('x', 'v', '1', 'failed', 'không đạt', 'false', 'fail'):
|
||
failed += 1
|
||
if is_mandatory:
|
||
mandatory_failed += 1
|
||
else:
|
||
optional_failed += 1
|
||
|
||
wb.close()
|
||
|
||
total = passed + failed
|
||
info['passed_count'] = passed
|
||
info['failed_count'] = failed
|
||
info['total_checks'] = total
|
||
info['compliance_percentage'] = round((passed / total) * 100, 1) if total > 0 else 0.0
|
||
info['mandatory_passed'] = mandatory_passed
|
||
info['mandatory_failed'] = mandatory_failed
|
||
info['mandatory_total'] = mandatory_passed + mandatory_failed
|
||
info['mandatory_percentage'] = round((mandatory_passed / info['mandatory_total']) * 100, 1) if info['mandatory_total'] > 0 else 0.0
|
||
info['optional_passed'] = optional_passed
|
||
info['optional_failed'] = optional_failed
|
||
info['optional_total'] = optional_passed + optional_failed
|
||
|
||
except Exception as e:
|
||
print(Fore.YELLOW + f"[REPORTS] Error parsing Excel {excel_file}: {e}" + Fore.RESET)
|
||
|
||
# Save to DB
|
||
new_record = models.AuditRecord(**info)
|
||
db.add(new_record)
|
||
|
||
if new_files:
|
||
db.commit()
|
||
print(Fore.GREEN + f"[DB SYNC] Synced {len(new_files)} missing files to database." + Fore.RESET)
|
||
|
||
# Backfill tiêu chí bắt buộc cho các bản ghi Excel cũ trong DB (nếu mandatory_total == 0)
|
||
unparsed_mand = db.query(models.AuditRecord).filter(models.AuditRecord.mandatory_total == 0).all()
|
||
if unparsed_mand:
|
||
up_mand = False
|
||
for rec in unparsed_mand:
|
||
if not rec.filename.endswith('.xlsx'):
|
||
continue
|
||
epath = os.path.join(base_dir, rec.filename)
|
||
if not os.path.exists(epath):
|
||
continue
|
||
try:
|
||
from openpyxl import load_workbook
|
||
wb_m = load_workbook(epath, read_only=True, data_only=True)
|
||
ws_m = wb_m.worksheets[0]
|
||
mp, mf = 0, 0
|
||
for row_m in ws_m.iter_rows(min_row=1, max_row=ws_m.max_row):
|
||
col_c = str(row_m[2].value).strip().lower() if len(row_m) > 2 and row_m[2].value else ''
|
||
col_d = str(row_m[3].value).strip().lower() if len(row_m) > 3 and row_m[3].value else ''
|
||
col_e = str(row_m[4].value).strip().lower() if len(row_m) > 4 and row_m[4].value else ''
|
||
if col_e in ('x', 'v', '1', 'có', 'true', 'yes', 'bắt buộc'):
|
||
if col_c in ('x', 'v', '1', 'passed', 'đạt', 'true', 'ok'):
|
||
mp += 1
|
||
elif col_d in ('x', 'v', '1', 'failed', 'không đạt', 'false', 'fail'):
|
||
mf += 1
|
||
wb_m.close()
|
||
if mp + mf > 0:
|
||
rec.mandatory_passed = mp
|
||
rec.mandatory_failed = mf
|
||
rec.mandatory_total = mp + mf
|
||
rec.mandatory_percentage = round((mp / (mp + mf)) * 100, 1)
|
||
up_mand = True
|
||
except Exception:
|
||
pass
|
||
if up_mand:
|
||
db.commit()
|
||
|
||
# Update mandatory_percentage for records with mandatory_total > 0
|
||
fix_mand_pct = db.query(models.AuditRecord).filter(
|
||
models.AuditRecord.mandatory_total > 0,
|
||
models.AuditRecord.mandatory_percentage == 0.0
|
||
).all()
|
||
if fix_mand_pct:
|
||
for rec in fix_mand_pct:
|
||
if rec.mandatory_total > 0:
|
||
rec.mandatory_percentage = round((rec.mandatory_passed / rec.mandatory_total) * 100, 1)
|
||
db.commit()
|
||
|
||
# Lấy thông tin đơn vị và người upload từ UserFileRecord
|
||
user_file_records = db.query(models.UserFileRecord).all()
|
||
user_file_map = {uf.filename: ("TSC" if uf.unit_name and uf.unit_name.strip().upper() == "SOFT" else uf.unit_name) for uf in user_file_records}
|
||
user_upload_map = {uf.filename: uf.username for uf in user_file_records}
|
||
|
||
# Query all records
|
||
records = db.query(models.AuditRecord).all()
|
||
valid_records = [r for r in records if r.filename in excel_files]
|
||
all_disk_txts = {f for f in excel_files if f.endswith('.txt')}
|
||
|
||
non_txt_records = [r for r in valid_records if not r.filename.endswith('.txt')]
|
||
txt_records = [r for r in valid_records if r.filename.endswith('.txt')]
|
||
|
||
claimed_txt_files = set()
|
||
seen_host_dates = {}
|
||
db_modified = False
|
||
db_rec_map = {r.filename: r for r in valid_records}
|
||
|
||
for r in non_txt_records:
|
||
has_txt = r.has_txt
|
||
txt_filename = r.txt_filename
|
||
base_name = r.filename.rsplit('.', 1)[0]
|
||
|
||
if txt_filename and txt_filename in all_disk_txts:
|
||
has_txt = True
|
||
elif (base_name + '.txt') in all_disk_txts:
|
||
has_txt = True
|
||
txt_filename = base_name + '.txt'
|
||
else:
|
||
for tf in all_disk_txts:
|
||
if r.hostname and r.hostname.lower().replace('-', '').replace('_', '') in tf.lower().replace('-', '').replace('_', ''):
|
||
has_txt = True
|
||
txt_filename = tf
|
||
break
|
||
|
||
if has_txt and txt_filename:
|
||
claimed_txt_files.add(txt_filename)
|
||
if not r.has_txt or r.txt_filename != txt_filename:
|
||
r.has_txt = True
|
||
r.txt_filename = txt_filename
|
||
db_modified = True
|
||
claimed_txt_files.add(base_name + '.txt')
|
||
|
||
item = {
|
||
'filename': r.filename,
|
||
'file_size': r.file_size,
|
||
'file_date': r.file_date,
|
||
'hostname': r.hostname,
|
||
'ip_address': r.ip_address,
|
||
'os_name': r.os_name,
|
||
'compliance_percentage': r.compliance_percentage,
|
||
'passed_count': r.passed_count,
|
||
'failed_count': r.failed_count,
|
||
'total_checks': r.total_checks,
|
||
'mandatory_passed': r.mandatory_passed,
|
||
'mandatory_failed': r.mandatory_failed,
|
||
'mandatory_total': r.mandatory_total,
|
||
'mandatory_percentage': r.mandatory_percentage,
|
||
'optional_passed': r.optional_passed,
|
||
'optional_failed': r.optional_failed,
|
||
'optional_total': r.optional_total,
|
||
'has_txt': has_txt,
|
||
'txt_filename': txt_filename,
|
||
'unit_name': user_file_map.get(r.filename, ''),
|
||
'uploaded_by': user_upload_map.get(r.filename, '')
|
||
}
|
||
results.append(item)
|
||
key = (r.hostname.strip().lower(), r.file_date) if r.hostname else None
|
||
if key:
|
||
seen_host_dates[key] = item
|
||
|
||
for r in txt_records:
|
||
if r.filename in claimed_txt_files:
|
||
continue
|
||
|
||
merged = False
|
||
for item in results:
|
||
ip_match = (item.get('ip_address') and r.ip_address and item['ip_address'].strip() != '-' and item['ip_address'].strip() == r.ip_address.strip())
|
||
host_match = (item.get('hostname') and r.hostname and item['hostname'].strip() != '-' and item['hostname'].strip().lower() == r.hostname.strip().lower())
|
||
date_match = (item.get('file_date', '')[:10] == (r.file_date or '')[:10])
|
||
|
||
base_item = re.sub(r'_\d{4}_\d{2}_\d{2}.*$', '', item['filename'].rsplit('.', 1)[0])
|
||
base_txt = re.sub(r'_\d{4}_\d{2}_\d{2}.*$', '', r.filename.rsplit('.', 1)[0])
|
||
prefix_match = (len(base_item) > 5 and base_item.lower() == base_txt.lower())
|
||
|
||
if (date_match and (ip_match or host_match)) or prefix_match:
|
||
if not item['has_txt'] or not item['txt_filename']:
|
||
item['has_txt'] = True
|
||
item['txt_filename'] = r.filename
|
||
main_rec = db_rec_map.get(item['filename'])
|
||
if main_rec:
|
||
main_rec.has_txt = True
|
||
main_rec.txt_filename = r.filename
|
||
if not item.get('unit_name') or item['unit_name'] == '-':
|
||
txt_unit = user_file_map.get(r.filename, '')
|
||
if txt_unit:
|
||
item['unit_name'] = txt_unit
|
||
if not item.get('uploaded_by') or item['uploaded_by'] == '-':
|
||
txt_user = user_upload_map.get(r.filename, '')
|
||
if txt_user:
|
||
item['uploaded_by'] = txt_user
|
||
merged = True
|
||
try:
|
||
db.delete(r)
|
||
db_modified = True
|
||
except Exception:
|
||
pass
|
||
break
|
||
|
||
if merged:
|
||
continue
|
||
|
||
results.append({
|
||
'filename': r.filename,
|
||
'file_size': r.file_size,
|
||
'file_date': r.file_date,
|
||
'hostname': r.hostname,
|
||
'ip_address': r.ip_address,
|
||
'os_name': r.os_name,
|
||
'compliance_percentage': r.compliance_percentage,
|
||
'passed_count': r.passed_count,
|
||
'failed_count': r.failed_count,
|
||
'total_checks': r.total_checks,
|
||
'mandatory_passed': r.mandatory_passed,
|
||
'mandatory_failed': r.mandatory_failed,
|
||
'mandatory_total': r.mandatory_total,
|
||
'mandatory_percentage': r.mandatory_percentage,
|
||
'optional_passed': r.optional_passed,
|
||
'optional_failed': r.optional_failed,
|
||
'optional_total': r.optional_total,
|
||
'has_txt': True,
|
||
'txt_filename': r.filename,
|
||
'unit_name': user_file_map.get(r.filename, ''),
|
||
'uploaded_by': user_upload_map.get(r.filename, '')
|
||
})
|
||
|
||
# Tự động đồng bộ đơn vị và thông số bắt buộc cho các file cùng hostname
|
||
host_units = {}
|
||
host_mands = {}
|
||
for item in results:
|
||
h = (item.get('hostname') or '').strip().lower()
|
||
if h:
|
||
if item.get('unit_name') and item['unit_name'] != '-':
|
||
host_units[h] = item['unit_name']
|
||
if item.get('mandatory_total', 0) > 0:
|
||
host_mands[h] = (item['mandatory_passed'], item['mandatory_failed'], item['mandatory_total'], item['mandatory_percentage'])
|
||
for item in results:
|
||
h = (item.get('hostname') or '').strip().lower()
|
||
if h:
|
||
if not item.get('unit_name') or item['unit_name'] == '-':
|
||
u = host_units.get(h, '')
|
||
if u:
|
||
item['unit_name'] = u
|
||
if item.get('mandatory_total', 0) == 0 and h in host_mands:
|
||
mp, mf, mt, mpct = host_mands[h]
|
||
item['mandatory_passed'] = mp
|
||
item['mandatory_failed'] = mf
|
||
item['mandatory_total'] = mt
|
||
item['mandatory_percentage'] = mpct
|
||
if item.get('mandatory_total', 0) > 0:
|
||
item['mandatory_percentage'] = round((item.get('mandatory_passed', 0) / item['mandatory_total']) * 100, 1)
|
||
|
||
if db_modified:
|
||
try:
|
||
db.commit()
|
||
except Exception as e:
|
||
db.rollback()
|
||
|
||
# Sort by file date (assuming DD/MM/YYYY HH:MM)
|
||
results.sort(key=lambda x: datetime.strptime(x['file_date'], '%d/%m/%Y %H:%M') if x['file_date'] else datetime.min, reverse=True)
|
||
return results
|
||
finally:
|
||
db.close()
|
||
|
||
|
||
@app.get("/api/admin/output_files")
|
||
async def api_admin_output_files(
|
||
request: Request,
|
||
q: str = ""
|
||
):
|
||
"""API: List and search output files with system info. Requires login (session auth)."""
|
||
user = get_session_user(request)
|
||
if not user:
|
||
raise HTTPException(status_code=401, detail="Authentication required")
|
||
|
||
try:
|
||
import asyncio
|
||
all_files = await asyncio.to_thread(extract_info_from_output_files)
|
||
|
||
# Filter by search query
|
||
if q:
|
||
q_lower = q.lower().strip()
|
||
filtered = []
|
||
for f in all_files:
|
||
if (q_lower in f.get('hostname','').lower() or
|
||
q_lower in f.get('ip_address','').lower() or
|
||
q_lower in f.get('filename','').lower() or
|
||
q_lower in f.get('os_name','').lower() or
|
||
q_lower in f.get('unit_name','').lower() or
|
||
q_lower in f.get('uploaded_by','').lower()):
|
||
filtered.append(f)
|
||
return filtered
|
||
|
||
return all_files
|
||
except Exception as e:
|
||
import traceback
|
||
error_info = traceback.format_exc()
|
||
print(Fore.RED + f"[API ERROR] api_admin_output_files failed: {str(e)}\n{error_info}" + Fore.RESET)
|
||
raise HTTPException(status_code=500, detail=f"Lỗi máy chủ: {str(e)}")
|
||
|
||
@app.post("/admin/reports/generate")
|
||
async def admin_generate_report(
|
||
request: Request
|
||
):
|
||
"""Generate Excel summary report from selected files."""
|
||
try:
|
||
body = await request.json()
|
||
selected_files = body.get('files', [])
|
||
except Exception:
|
||
selected_files = []
|
||
|
||
# Thử session auth trước
|
||
user = get_session_user(request)
|
||
|
||
# Fallback sang HTTP Basic auth
|
||
if not user:
|
||
auth_header = request.headers.get('Authorization')
|
||
if auth_header:
|
||
try:
|
||
scheme, credentials_str = auth_header.split()
|
||
if scheme.lower() == 'basic':
|
||
decoded = base64.b64decode(credentials_str).decode('utf-8')
|
||
username, password = decoded.split(':', 1)
|
||
credentials = HTTPBasicCredentials(username=username, password=password)
|
||
user = verify_credentials(credentials)
|
||
except Exception:
|
||
pass
|
||
|
||
# Fallback cho guest session trong tab File của tôi
|
||
if not user:
|
||
session_id = request.cookies.get('session_id')
|
||
session_files = sessions.get(session_id, {}).get('files', []) if session_id else []
|
||
if selected_files and set(selected_files).issubset(set(session_files)):
|
||
user = {"username": "guest", "role": "guest"}
|
||
|
||
if not user:
|
||
raise HTTPException(status_code=401, detail="Authentication required")
|
||
|
||
try:
|
||
if not selected_files:
|
||
raise HTTPException(status_code=400, detail="No files selected")
|
||
|
||
# Get info for selected files
|
||
all_files = extract_info_from_output_files()
|
||
selected_info = [f for f in all_files if f['filename'] in selected_files]
|
||
|
||
if not selected_info:
|
||
raise HTTPException(status_code=400, detail="No matching files found")
|
||
|
||
# Generate Excel summary
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
|
||
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "Tổng hợp kết quả"
|
||
|
||
# Styles
|
||
header_font = Font(name='Segoe UI', size=11, bold=True, color='FFFFFF')
|
||
header_fill = PatternFill(start_color='1565C0', end_color='1976D2', fill_type='solid')
|
||
header_align = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||
cell_font = Font(name='Segoe UI', size=10)
|
||
cell_align = Alignment(vertical='center', wrap_text=True)
|
||
center_align = Alignment(horizontal='center', vertical='center')
|
||
thin_border = Border(
|
||
left=Side(style='thin'),
|
||
right=Side(style='thin'),
|
||
top=Side(style='thin'),
|
||
bottom=Side(style='thin')
|
||
)
|
||
green_fill = PatternFill(start_color='E8F5E9', fill_type='solid')
|
||
yellow_fill = PatternFill(start_color='FFF3E0', fill_type='solid')
|
||
red_fill = PatternFill(start_color='FFEBEE', fill_type='solid')
|
||
|
||
# Title row
|
||
ws.merge_cells('A1:H1')
|
||
title_cell = ws['A1']
|
||
title_cell.value = f"BÁO CÁO TỔNG HỢP KẾT QUẢ KIỂM TRA HARDENING"
|
||
title_cell.font = Font(name='Segoe UI', size=14, bold=True, color='1565C0')
|
||
title_cell.alignment = Alignment(horizontal='center', vertical='center')
|
||
|
||
ws.merge_cells('A2:H2')
|
||
date_cell = ws['A2']
|
||
date_cell.value = f"Ngày xuất: {datetime.now().strftime('%d/%m/%Y %H:%M')}"
|
||
date_cell.font = Font(name='Segoe UI', size=10, italic=True, color='666666')
|
||
date_cell.alignment = Alignment(horizontal='center')
|
||
|
||
# Headers
|
||
headers = ['STT', 'Tên máy chủ', 'Hệ điều hành', 'Địa chỉ IP',
|
||
'Tỉ lệ đạt (%)', 'Bắt buộc đạt', 'TC bắt buộc chưa đạt', 'Ghi chú', 'Người upload']
|
||
col_widths = [6, 30, 22, 18, 14, 14, 20, 35, 20]
|
||
|
||
for col_idx, (header, width) in enumerate(zip(headers, col_widths), 1):
|
||
cell = ws.cell(row=4, column=col_idx, value=header)
|
||
cell.font = header_font
|
||
cell.fill = header_fill
|
||
cell.alignment = header_align
|
||
cell.border = thin_border
|
||
ws.column_dimensions[cell.column_letter].width = width
|
||
|
||
# Data rows
|
||
for row_idx, info in enumerate(selected_info, 5):
|
||
stt = row_idx - 4
|
||
|
||
ws.cell(row=row_idx, column=1, value=stt).alignment = center_align
|
||
ws.cell(row=row_idx, column=2, value=info['hostname'])
|
||
ws.cell(row=row_idx, column=3, value=info['os_name'])
|
||
ws.cell(row=row_idx, column=4, value=info['ip_address'])
|
||
|
||
pct_cell = ws.cell(row=row_idx, column=5, value=info['compliance_percentage'])
|
||
pct_cell.alignment = center_align
|
||
pct_cell.number_format = '0.0'
|
||
|
||
mand_cell = ws.cell(row=row_idx, column=6,
|
||
value=f"{info['mandatory_passed']}/{info['mandatory_total']}")
|
||
mand_cell.alignment = center_align
|
||
|
||
ws.cell(row=row_idx, column=7, value=info['mandatory_failed']).alignment = center_align
|
||
|
||
# Note
|
||
note = ''
|
||
if info['mandatory_failed'] > 0 and info['compliance_percentage'] < 80:
|
||
note = f"Còn {info['mandatory_failed']} tiêu chí bắt buộc chưa đạt. Tỉ lệ đạt chỉ {info['compliance_percentage']}%"
|
||
elif info['mandatory_failed'] > 0:
|
||
note = f"Còn {info['mandatory_failed']} tiêu chí bắt buộc chưa đạt"
|
||
elif info['compliance_percentage'] < 80:
|
||
note = f"Tỉ lệ đạt chỉ {info['compliance_percentage']}% (yêu cầu ≥80%)"
|
||
else:
|
||
note = "Đạt yêu cầu"
|
||
ws.cell(row=row_idx, column=8, value=note)
|
||
|
||
# Uploader
|
||
ws.cell(row=row_idx, column=9, value=info.get('uploaded_by', ''))
|
||
|
||
# Color coding
|
||
if info['compliance_percentage'] >= 80 and info['mandatory_failed'] == 0:
|
||
row_fill = green_fill
|
||
elif info['compliance_percentage'] >= 50:
|
||
row_fill = yellow_fill
|
||
else:
|
||
row_fill = red_fill
|
||
|
||
for col in range(1, 10):
|
||
cell = ws.cell(row=row_idx, column=col)
|
||
cell.font = cell_font
|
||
cell.border = thin_border
|
||
cell.fill = row_fill
|
||
if not cell.alignment or cell.alignment.horizontal is None:
|
||
cell.alignment = cell_align
|
||
|
||
# Summary row
|
||
summary_row = 5 + len(selected_info)
|
||
ws.merge_cells(f'A{summary_row}:D{summary_row}')
|
||
summary_label = ws.cell(row=summary_row, column=1, value=f"TỔNG HỢP ({len(selected_info)} máy chủ)")
|
||
summary_label.font = Font(name='Segoe UI', size=10, bold=True)
|
||
summary_label.alignment = Alignment(horizontal='right', vertical='center')
|
||
summary_label.border = thin_border
|
||
for c in range(2, 5):
|
||
ws.cell(row=summary_row, column=c).border = thin_border
|
||
|
||
total_passed = sum(f['passed_count'] for f in selected_info)
|
||
total_checks = sum(f['total_checks'] for f in selected_info)
|
||
avg_pct = round((total_passed / total_checks) * 100, 1) if total_checks > 0 else 0
|
||
|
||
avg_cell = ws.cell(row=summary_row, column=5, value=avg_pct)
|
||
avg_cell.font = Font(name='Segoe UI', size=10, bold=True)
|
||
avg_cell.alignment = center_align
|
||
avg_cell.border = thin_border
|
||
avg_cell.number_format = '0.0'
|
||
|
||
total_mp = sum(f['mandatory_passed'] for f in selected_info)
|
||
total_mt = sum(f['mandatory_total'] for f in selected_info)
|
||
ws.cell(row=summary_row, column=6, value=f"{total_mp}/{total_mt}").font = Font(name='Segoe UI', size=10, bold=True)
|
||
ws.cell(row=summary_row, column=6).alignment = center_align
|
||
ws.cell(row=summary_row, column=6).border = thin_border
|
||
|
||
total_mf = sum(f['mandatory_failed'] for f in selected_info)
|
||
ws.cell(row=summary_row, column=7, value=total_mf).font = Font(name='Segoe UI', size=10, bold=True)
|
||
ws.cell(row=summary_row, column=7).alignment = center_align
|
||
ws.cell(row=summary_row, column=7).border = thin_border
|
||
|
||
ws.cell(row=summary_row, column=8).border = thin_border
|
||
ws.cell(row=summary_row, column=9).border = thin_border
|
||
|
||
# Save to output folder
|
||
report_filename = f"Tong_hop_hardening_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
|
||
report_path = os.path.join(OUTPUT_FOLDER, report_filename)
|
||
wb.save(report_path)
|
||
wb.close()
|
||
|
||
print(Fore.GREEN + f"[REPORTS] Generated summary report: {report_filename}" + Fore.RESET)
|
||
|
||
root_path = get_app_root_path(request)
|
||
return {
|
||
"success": True,
|
||
"filename": report_filename,
|
||
"download_url": f"{root_path}/download/{report_filename}",
|
||
"count": len(selected_info)
|
||
}
|
||
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
import traceback
|
||
print(Fore.RED + f"[REPORTS] Error generating report: {e}" + Fore.RESET)
|
||
print(traceback.format_exc())
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.post("/admin/reports/generate-billing")
|
||
async def admin_generate_billing_report(request: Request):
|
||
"""Generate Excel billing report (Thống kê sản lượng)."""
|
||
user = get_session_user(request)
|
||
if not user:
|
||
raise HTTPException(status_code=401, detail="Authentication required")
|
||
|
||
try:
|
||
body = await request.json()
|
||
unit_price = body.get('unit_price', 2000000)
|
||
units = body.get('units', [])
|
||
months = body.get('months', [])
|
||
year_str = body.get('year', '')
|
||
except Exception:
|
||
raise HTTPException(status_code=400, detail="Invalid request body")
|
||
|
||
try:
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
|
||
from openpyxl.utils import get_column_letter
|
||
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "Sản lượng Hardening"
|
||
|
||
title_font = Font(name='Times New Roman', size=16, bold=True)
|
||
sub_font = Font(name='Times New Roman', size=12, bold=False)
|
||
price_font = Font(name='Times New Roman', size=12, bold=True)
|
||
header_font = Font(name='Times New Roman', size=12, bold=True)
|
||
data_font = Font(name='Times New Roman', size=12)
|
||
bold_font = Font(name='Times New Roman', size=12, bold=True)
|
||
|
||
center_align = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||
left_align = Alignment(horizontal='left', vertical='center')
|
||
right_align = Alignment(horizontal='right', vertical='center')
|
||
|
||
thin = Side(border_style="thin", color="000000")
|
||
thin_border = Border(left=thin, right=thin, top=thin, bottom=thin)
|
||
|
||
total_cols = 2 + max(len(months), 2) + 1
|
||
last_col_letter = get_column_letter(total_cols)
|
||
|
||
ws.column_dimensions['A'].width = 8
|
||
ws.column_dimensions['B'].width = 28
|
||
for idx in range(max(len(months), 2)):
|
||
ws.column_dimensions[get_column_letter(3 + idx)].width = 18
|
||
ws.column_dimensions[last_col_letter].width = 18
|
||
|
||
# Row 1: Title
|
||
ws.merge_cells(f'A1:{last_col_letter}1')
|
||
t_cell = ws['A1']
|
||
t_cell.value = "3.3. Dịch vụ kiểm định hệ thống" + (f" (Năm {year_str})" if year_str else "")
|
||
t_cell.font = title_font
|
||
t_cell.alignment = center_align
|
||
ws.row_dimensions[1].height = 35
|
||
|
||
# Row 2: Unit Price
|
||
ws.merge_cells('A2:B2')
|
||
p_label = ws['A2']
|
||
p_label.value = "Đơn giá /thiết bị (VNĐ):"
|
||
p_label.font = sub_font
|
||
p_label.alignment = left_align
|
||
|
||
p_val = ws['C2']
|
||
p_val.value = int(unit_price)
|
||
p_val.font = price_font
|
||
p_val.number_format = '#,##0'
|
||
p_val.alignment = center_align
|
||
|
||
ws.row_dimensions[2].height = 25
|
||
ws.row_dimensions[3].height = 15
|
||
|
||
# Row 4 & 5: Headers
|
||
ws.merge_cells('A4:A5')
|
||
ws['A4'] = "STT"
|
||
ws.merge_cells('B4:B5')
|
||
ws['B4'] = "Tên đơn vị"
|
||
|
||
start_m_letter = get_column_letter(3)
|
||
end_m_letter = get_column_letter(2 + max(len(months), 2))
|
||
ws.merge_cells(f'{start_m_letter}4:{end_m_letter}4')
|
||
ws[f'{start_m_letter}4'] = "Sản lượng Hardening theo tháng"
|
||
|
||
display_months = months if len(months) >= 2 else (months + [""] * (2 - len(months)))
|
||
for idx, m in enumerate(display_months):
|
||
col_letter = get_column_letter(3 + idx)
|
||
ws[f'{col_letter}5'] = f"Tháng {m}" if m else ""
|
||
|
||
ws.merge_cells(f'{last_col_letter}4:{last_col_letter}5')
|
||
ws[f'{last_col_letter}4'] = "Tổng cộng"
|
||
|
||
for r_idx in (4, 5):
|
||
for c_idx in range(1, total_cols + 1):
|
||
cell = ws.cell(row=r_idx, column=c_idx)
|
||
cell.font = header_font
|
||
cell.alignment = center_align
|
||
cell.border = thin_border
|
||
ws.row_dimensions[4].height = 24
|
||
ws.row_dimensions[5].height = 24
|
||
|
||
# Data rows
|
||
current_row = 6
|
||
filtered_units = [u for u in units if u.get('unit_name', '').strip().upper() != 'ANTT']
|
||
for idx, u in enumerate(filtered_units):
|
||
u['stt'] = idx + 1
|
||
|
||
for item in filtered_units:
|
||
ws.cell(row=current_row, column=1, value=item.get('stt', '')).alignment = center_align
|
||
ws.cell(row=current_row, column=2, value=item.get('unit_name', '')).alignment = left_align
|
||
|
||
counts = item.get('counts', {})
|
||
for idx, m in enumerate(display_months):
|
||
c_val = counts.get(m, 0) if m else 0
|
||
cell = ws.cell(row=current_row, column=3 + idx, value=(c_val if c_val > 0 else ""))
|
||
cell.alignment = right_align
|
||
if c_val > 0:
|
||
cell.number_format = '#,##0'
|
||
|
||
tot_cell = ws.cell(row=current_row, column=total_cols, value=f"=SUM({start_m_letter}{current_row}:{end_m_letter}{current_row})")
|
||
tot_cell.alignment = right_align
|
||
tot_cell.number_format = '#,##0'
|
||
|
||
for col_idx in range(1, total_cols + 1):
|
||
c = ws.cell(row=current_row, column=col_idx)
|
||
c.font = data_font
|
||
c.border = thin_border
|
||
|
||
ws.row_dimensions[current_row].height = 22
|
||
current_row += 1
|
||
|
||
# Total Devices Row
|
||
ws.cell(row=current_row, column=1, value="")
|
||
ws.cell(row=current_row, column=2, value="Tổng (thiết bị)").alignment = left_align
|
||
for idx in range(len(display_months)):
|
||
col_letter = get_column_letter(3 + idx)
|
||
c = ws.cell(row=current_row, column=3 + idx, value=f"=SUM({col_letter}6:{col_letter}{current_row-1})")
|
||
c.alignment = right_align
|
||
c.number_format = '#,##0'
|
||
|
||
tot_grand = ws.cell(row=current_row, column=total_cols, value=f"=SUM({last_col_letter}6:{last_col_letter}{current_row-1})")
|
||
tot_grand.alignment = right_align
|
||
tot_grand.number_format = '#,##0'
|
||
|
||
for col_idx in range(1, total_cols + 1):
|
||
c = ws.cell(row=current_row, column=col_idx)
|
||
c.font = bold_font
|
||
c.border = thin_border
|
||
ws.row_dimensions[current_row].height = 25
|
||
current_row += 1
|
||
|
||
# Total Cost Row
|
||
ws.cell(row=current_row, column=1, value="")
|
||
ws.cell(row=current_row, column=2, value="Thành tiền (VNĐ)").alignment = left_align
|
||
for idx in range(len(display_months)):
|
||
col_letter = get_column_letter(3 + idx)
|
||
c = ws.cell(row=current_row, column=3 + idx, value=f"={col_letter}{current_row-1}*$C$2")
|
||
c.alignment = right_align
|
||
c.number_format = '#,##0'
|
||
|
||
cost_grand = ws.cell(row=current_row, column=total_cols, value=f"={last_col_letter}{current_row-1}*$C$2")
|
||
cost_grand.alignment = right_align
|
||
cost_grand.number_format = '#,##0'
|
||
|
||
for col_idx in range(1, total_cols + 1):
|
||
c = ws.cell(row=current_row, column=col_idx)
|
||
c.font = bold_font
|
||
c.border = thin_border
|
||
ws.row_dimensions[current_row].height = 25
|
||
|
||
report_filename = f"Thong_ke_san_luong_hardening_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
|
||
report_path = os.path.join(OUTPUT_FOLDER, report_filename)
|
||
wb.save(report_path)
|
||
wb.close()
|
||
|
||
print(Fore.GREEN + f"[REPORTS] Generated billing report: {report_filename}" + Fore.RESET)
|
||
|
||
root_path = get_app_root_path(request)
|
||
return {
|
||
"success": True,
|
||
"filename": report_filename,
|
||
"download_url": f"{root_path}/download/{report_filename}"
|
||
}
|
||
except Exception as e:
|
||
import traceback
|
||
print(Fore.RED + f"[BILLING REPORT ERROR]: {e}\n{traceback.format_exc()}" + Fore.RESET)
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
# ============== API ENDPOINTS FOR REMOTE SERVERS ==============
|
||
|
||
def load_api_settings():
|
||
"""Load API key and allowed IPs from users_config.json"""
|
||
with open('users_config.json', "r", encoding="utf-8") as f:
|
||
data = json.load(f)
|
||
settings = data.get("settings", {})
|
||
api_key = settings.get("api_key")
|
||
allowed_ips = settings.get("allowed_ips", [])
|
||
return api_key, allowed_ips
|
||
|
||
def check_ip_allowed(client_ip: str, allowed_ips: list) -> bool:
|
||
"""Check if client IP is in the allowed list (supports CIDR notation)"""
|
||
for allowed in allowed_ips:
|
||
# Exact match
|
||
if allowed == client_ip:
|
||
return True
|
||
# CIDR subnet
|
||
if "/" in allowed:
|
||
try:
|
||
if ipaddress.ip_address(client_ip) in ipaddress.ip_network(allowed, strict=False):
|
||
return True
|
||
except ValueError:
|
||
continue
|
||
return False
|
||
|
||
def get_client_ip(request: Request) -> str:
|
||
"""
|
||
Lấy IP client thực sự, ưu tiên X-Forwarded-For do reverse proxy set.
|
||
Nếu không có thì fallback về request.client.host.
|
||
"""
|
||
xff = request.headers.get("x-forwarded-for")
|
||
if xff:
|
||
# Thường dạng "ip_goc, ip_proxy1, ip_proxy2"
|
||
return xff.split(",")[0].strip()
|
||
return request.client.host
|
||
|
||
@app.post("/api/upload")
|
||
async def api_upload_files(
|
||
request: Request,
|
||
os_type: str = Form(...),
|
||
api_key: str = Form(...),
|
||
encrypted_files: List[UploadFile] = File(...)
|
||
):
|
||
"""
|
||
API endpoint for uploading and processing encrypted files from remote servers.
|
||
Requires API key authentication and IP whitelist.
|
||
Returns JSON response with processed file information.
|
||
"""
|
||
# >>> AUTH <<<
|
||
stored_api_key, allowed_ips = load_api_settings()
|
||
|
||
# Check API key
|
||
if api_key != stored_api_key:
|
||
print(Fore.RED + f"[API] Invalid API key attempt" + Fore.RESET)
|
||
raise HTTPException(status_code=401, detail="Invalid API Key")
|
||
|
||
# Check IP (lấy IP thật từ X-Forwarded-For)
|
||
client_ip = get_client_ip(request)
|
||
if not check_ip_allowed(client_ip, allowed_ips):
|
||
print(Fore.RED + f"[API] IP not allowed: {client_ip}" + Fore.RESET)
|
||
raise HTTPException(status_code=403, detail=f"IP not allowed: {client_ip}")
|
||
|
||
print(Fore.GREEN + f"[API] Authorized request from IP: {client_ip}" + Fore.RESET)
|
||
|
||
try:
|
||
# Auto-detect OS from filename if os_type is 'auto'
|
||
if not os_type or os_type == 'auto':
|
||
filenames = [f.filename for f in encrypted_files if f.filename]
|
||
detected_os, _ = detect_os_from_multiple_files(filenames)
|
||
if detected_os:
|
||
os_type = detected_os
|
||
print(Fore.GREEN + f"[API] Auto-detected OS: {detected_os}" + Fore.RESET)
|
||
else:
|
||
raise HTTPException(status_code=400, detail="Cannot auto-detect OS. Please specify os_type.")
|
||
|
||
# Validate OS type
|
||
if os_type not in OS_CONFIG:
|
||
raise HTTPException(status_code=400, detail=f"Invalid os_type: {os_type}")
|
||
|
||
# Validate file exists
|
||
if not encrypted_files:
|
||
raise HTTPException(status_code=400, detail="No encrypted files uploaded")
|
||
|
||
os_info = OS_CONFIG[os_type]
|
||
checklist_path = os.path.join(CONFIG_FOLDER, os_info['checklist'])
|
||
config_path = os.path.join(CONFIG_FOLDER, os_info['config'])
|
||
|
||
# Verify exists
|
||
if not os.path.exists(checklist_path):
|
||
raise HTTPException(status_code=400, detail=f"Checklist not found: {os_info['checklist']}")
|
||
|
||
if not os.path.exists(config_path):
|
||
raise HTTPException(status_code=400, detail=f"Config not found: {os_info['config']}")
|
||
|
||
output_files = []
|
||
system_info_list = []
|
||
|
||
for enc_file in encrypted_files:
|
||
if enc_file.filename and allowed_file(enc_file.filename):
|
||
enc_filename = secure_filename(enc_file.filename)
|
||
enc_path = os.path.join(UPLOAD_FOLDER, enc_filename)
|
||
|
||
# Save uploaded file
|
||
with open(enc_path, "wb") as buffer:
|
||
buffer.write(await enc_file.read())
|
||
|
||
print(Fore.BLUE + f"[API] Processing: {enc_filename}" + Fore.RESET)
|
||
original_dir = os.getcwd()
|
||
|
||
try:
|
||
# Run decrypt + report
|
||
system_info = audit_decryption.run_generate_excel(checklist_path, config_path, enc_path)
|
||
if system_info:
|
||
system_info_list.append(system_info)
|
||
|
||
# Move decrypted
|
||
dec_filename = enc_filename.replace('.enc', '')
|
||
if os.path.exists(dec_filename):
|
||
output_dec_path = os.path.join(OUTPUT_FOLDER, dec_filename)
|
||
if os.path.exists(output_dec_path):
|
||
os.remove(output_dec_path)
|
||
shutil.move(dec_filename, output_dec_path)
|
||
|
||
# Search excel output
|
||
search_dirs = ['.', UPLOAD_FOLDER, os.path.dirname(__file__)]
|
||
excel_found = False
|
||
|
||
for search_dir in search_dirs:
|
||
if not os.path.exists(search_dir):
|
||
continue
|
||
|
||
for file in os.listdir(search_dir):
|
||
if file.endswith('.xlsx') and 'Checklist' not in file:
|
||
source_path = os.path.join(search_dir, file)
|
||
output_path = os.path.join(OUTPUT_FOLDER, file)
|
||
|
||
if os.path.exists(output_path):
|
||
os.remove(output_path)
|
||
|
||
shutil.move(source_path, output_path)
|
||
output_files.append(file)
|
||
excel_found = True
|
||
print(Fore.GREEN + f"[API] Generated: {file}" + Fore.RESET)
|
||
break
|
||
|
||
if excel_found:
|
||
break
|
||
|
||
except Exception as e:
|
||
print(Fore.RED + f"[API] Error processing {enc_filename}: {str(e)}" + Fore.RESET)
|
||
finally:
|
||
os.chdir(original_dir)
|
||
|
||
print(Fore.GREEN + f"[API] Successfully processed {len(output_files)} file(s)" + Fore.RESET)
|
||
return {
|
||
"success": True,
|
||
"message": "Processing complete",
|
||
"processed": len(output_files),
|
||
"output_files": output_files,
|
||
"os_type": os_type,
|
||
"download_urls": [f"/api/download/{f}" for f in output_files]
|
||
}
|
||
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
print(Fore.RED + f"[API] Error: {str(e)}" + Fore.RESET)
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.post("/api/download")
|
||
async def api_download_file(request: Request):
|
||
"""
|
||
API endpoint for downloading processed files.
|
||
Requires API key authentication and IP whitelist.
|
||
Accepts JSON body with filename and api_key.
|
||
"""
|
||
try:
|
||
# Parse JSON body
|
||
try:
|
||
body = await request.json()
|
||
except:
|
||
raise HTTPException(status_code=400, detail="Body must be JSON")
|
||
|
||
filename = body.get("filename")
|
||
api_key = body.get("api_key")
|
||
|
||
if not filename:
|
||
raise HTTPException(status_code=400, detail="Missing filename")
|
||
|
||
if not api_key:
|
||
raise HTTPException(status_code=400, detail="Missing api_key")
|
||
|
||
# Load API key + whitelist
|
||
stored_api_key, allowed_ips = load_api_settings()
|
||
|
||
# Check API key
|
||
if api_key != stored_api_key:
|
||
raise HTTPException(status_code=401, detail="Invalid API Key")
|
||
|
||
# Check whitelist IP (lấy IP thật từ X-Forwarded-For)
|
||
client_ip = get_client_ip(request)
|
||
if not check_ip_allowed(client_ip, allowed_ips):
|
||
raise HTTPException(status_code=403, detail=f"IP not allowed: {client_ip}")
|
||
|
||
# 🔒 SECURITY CHECK: Prevent directory traversal
|
||
normalized = os.path.realpath(os.path.join(OUTPUT_FOLDER, filename))
|
||
allowed_root = os.path.realpath(OUTPUT_FOLDER)
|
||
|
||
if not normalized.startswith(allowed_root):
|
||
raise HTTPException(status_code=403, detail="Access denied")
|
||
|
||
# Also ensure filename does not contain path separators
|
||
if "/" in filename or "\\" in filename or ".." in filename:
|
||
raise HTTPException(status_code=400, detail="Invalid filename")
|
||
|
||
# File must exist
|
||
if not os.path.exists(normalized):
|
||
raise HTTPException(status_code=404, detail="File not found")
|
||
|
||
print(Fore.GREEN + f"[API] Download: {filename} by {client_ip}" + Fore.RESET)
|
||
|
||
# Return file
|
||
return FileResponse(
|
||
normalized,
|
||
media_type='application/octet-stream',
|
||
filename=os.path.basename(normalized)
|
||
)
|
||
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.get("/api/download/{filename}")
|
||
async def api_download_file_get(
|
||
request: Request,
|
||
filename: str,
|
||
api_key: str = None
|
||
):
|
||
"""
|
||
API endpoint for downloading processed files via GET request.
|
||
Requires API key as query parameter and IP whitelist.
|
||
"""
|
||
try:
|
||
if not api_key:
|
||
raise HTTPException(status_code=400, detail="Missing api_key query parameter")
|
||
|
||
# Load API key + whitelist
|
||
stored_api_key, allowed_ips = load_api_settings()
|
||
|
||
# Check API key
|
||
if api_key != stored_api_key:
|
||
raise HTTPException(status_code=401, detail="Invalid API Key")
|
||
|
||
# Check whitelist IP
|
||
client_ip = get_client_ip(request)
|
||
if not check_ip_allowed(client_ip, allowed_ips):
|
||
raise HTTPException(status_code=403, detail=f"IP not allowed: {client_ip}")
|
||
|
||
# 🔒 SECURITY CHECK: Prevent directory traversal
|
||
normalized = os.path.realpath(os.path.join(OUTPUT_FOLDER, filename))
|
||
allowed_root = os.path.realpath(OUTPUT_FOLDER)
|
||
|
||
if not normalized.startswith(allowed_root):
|
||
raise HTTPException(status_code=403, detail="Access denied")
|
||
|
||
if "/" in filename or "\\" in filename or ".." in filename:
|
||
raise HTTPException(status_code=400, detail="Invalid filename")
|
||
|
||
if not os.path.exists(normalized):
|
||
raise HTTPException(status_code=404, detail="File not found")
|
||
|
||
print(Fore.GREEN + f"[API] Download: {filename} by {client_ip}" + Fore.RESET)
|
||
|
||
return FileResponse(
|
||
normalized,
|
||
media_type='application/octet-stream',
|
||
filename=os.path.basename(normalized)
|
||
)
|
||
|
||
except HTTPException:
|
||
raise
|
||
except Exception as e:
|
||
raise HTTPException(status_code=500, detail=str(e))
|
||
|
||
@app.get("/server_stats", response_class=HTMLResponse, name="server_stats_page")
|
||
async def server_stats_page(request: Request):
|
||
"""Render the server statistics dashboard."""
|
||
is_logged_in = request.cookies.get("session_id") in sessions
|
||
if not is_logged_in:
|
||
return RedirectResponse(url=request.url_for('login'), status_code=303)
|
||
|
||
session_id = request.cookies.get("session_id")
|
||
session_data = sessions.get(session_id, {})
|
||
user_info = get_session_user(request)
|
||
session_mode = session_data.get('session_mode', True)
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
# Đồng bộ dữ liệu từ disk vào DB trước khi thống kê
|
||
try:
|
||
extract_info_from_output_files()
|
||
except Exception as e:
|
||
print(f"[STATS SYNC] {e}")
|
||
|
||
records = db.query(models.AuditRecord).all()
|
||
|
||
# Fetch user file records to map filename to unit_name
|
||
user_file_records = db.query(models.UserFileRecord).all()
|
||
user_file_map = {uf.filename: ("TSC" if uf.unit_name and uf.unit_name.strip().upper() == "SOFT" else uf.unit_name) for uf in user_file_records}
|
||
user_upload_map = {uf.filename: uf.username for uf in user_file_records}
|
||
|
||
# Group by hostname/ip
|
||
hosts_data = {}
|
||
for r in records:
|
||
host = r.hostname.strip() if r.hostname else ""
|
||
if not host:
|
||
host = r.ip_address.strip() if r.ip_address else r.filename
|
||
|
||
if host not in hosts_data:
|
||
hosts_data[host] = []
|
||
|
||
hosts_data[host].append({
|
||
'date': r.file_date,
|
||
'os': r.os_name,
|
||
'ip': r.ip_address,
|
||
'compliance': r.compliance_percentage,
|
||
'passed': r.passed_count,
|
||
'failed': r.failed_count,
|
||
'total': r.total_checks,
|
||
'mandatory_percentage': r.mandatory_percentage,
|
||
'mandatory_passed': r.mandatory_passed,
|
||
'mandatory_total': r.mandatory_total,
|
||
'unit_name': user_file_map.get(r.filename, ''),
|
||
'uploaded_by': user_upload_map.get(r.filename, '')
|
||
})
|
||
|
||
# Đồng bộ đơn vị và tiêu chí bắt buộc giữa các lần quét của cùng 1 máy chủ
|
||
for host, points in hosts_data.items():
|
||
valid_unit = ""
|
||
valid_mp, valid_mt, valid_mpct = None, None, None
|
||
for pt in reversed(points):
|
||
if pt.get('unit_name') and pt['unit_name'] != '-':
|
||
valid_unit = pt['unit_name']
|
||
if pt.get('mandatory_total') and pt['mandatory_total'] > 0:
|
||
valid_mp = pt['mandatory_passed']
|
||
valid_mt = pt['mandatory_total']
|
||
valid_mpct = pt.get('mandatory_percentage')
|
||
if not valid_mpct or valid_mpct == 0:
|
||
valid_mpct = round((valid_mp / valid_mt) * 100, 1)
|
||
for pt in points:
|
||
if valid_unit and (not pt.get('unit_name') or pt['unit_name'] == '-'):
|
||
pt['unit_name'] = valid_unit
|
||
if valid_mt and (not pt.get('mandatory_total') or pt['mandatory_total'] == 0):
|
||
pt['mandatory_passed'] = valid_mp
|
||
pt['mandatory_total'] = valid_mt
|
||
pt['mandatory_percentage'] = valid_mpct if valid_mpct else round((valid_mp / valid_mt) * 100, 1)
|
||
if pt.get('mandatory_total', 0) > 0:
|
||
pt['mandatory_percentage'] = round((pt.get('mandatory_passed', 0) / pt['mandatory_total']) * 100, 1)
|
||
|
||
# Ensure chronological order
|
||
for host in hosts_data:
|
||
hosts_data[host].sort(key=lambda x: datetime.strptime(x['date'], '%d/%m/%Y %H:%M') if x['date'] else datetime.min)
|
||
|
||
return templates.TemplateResponse(
|
||
"server_stats.html",
|
||
{
|
||
"request": request,
|
||
"is_logged_in": True,
|
||
"user_info": user_info,
|
||
"session_mode": session_mode,
|
||
"hosts_data": json.dumps(hosts_data), # Make json available strictly mapped
|
||
"hosts_list": list(hosts_data.keys()),
|
||
"page_name": "admin",
|
||
"admin_tab": "stats"
|
||
}
|
||
)
|
||
finally:
|
||
db.close()
|
||
|
||
@app.get("/export_server_stats")
|
||
async def export_server_stats(request: Request, start_date: str = None, end_date: str = None, unit: str = None):
|
||
"""Export the grouped server stats to Excel."""
|
||
is_logged_in = request.cookies.get("session_id") in sessions
|
||
if not is_logged_in:
|
||
raise HTTPException(status_code=401, detail="Authentication required")
|
||
|
||
import io
|
||
from openpyxl import Workbook
|
||
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
|
||
|
||
db = SessionLocal()
|
||
try:
|
||
records = db.query(models.AuditRecord).all()
|
||
|
||
# Parse dates
|
||
start_dt = None
|
||
end_dt = None
|
||
if start_date:
|
||
try:
|
||
start_dt = datetime.strptime(start_date, '%d-%m-%Y')
|
||
except ValueError:
|
||
pass
|
||
if end_date:
|
||
try:
|
||
end_dt = datetime.strptime(end_date, '%d-%m-%Y')
|
||
end_dt = end_dt.replace(hour=23, minute=59, second=59)
|
||
except ValueError:
|
||
pass
|
||
|
||
# Fetch user file records to map filename to unit_name
|
||
user_file_records = db.query(models.UserFileRecord).all()
|
||
user_file_map = {uf.filename: ("TSC" if uf.unit_name and uf.unit_name.strip().upper() == "SOFT" else uf.unit_name) for uf in user_file_records}
|
||
user_upload_map = {uf.filename: uf.username for uf in user_file_records}
|
||
|
||
hosts_data = {}
|
||
for r in records:
|
||
# Filter by date
|
||
if r.file_date:
|
||
try:
|
||
r_dt = datetime.strptime(r.file_date, '%d/%m/%Y %H:%M')
|
||
if start_dt and r_dt < start_dt:
|
||
continue
|
||
if end_dt and r_dt > end_dt:
|
||
continue
|
||
except ValueError:
|
||
continue
|
||
else:
|
||
continue
|
||
|
||
unit_name = user_file_map.get(r.filename, '')
|
||
if unit and unit_name != unit:
|
||
continue
|
||
|
||
host = r.hostname.strip() if r.hostname else ""
|
||
if not host:
|
||
host = r.ip_address.strip() if r.ip_address else r.filename
|
||
|
||
if host not in hosts_data:
|
||
hosts_data[host] = []
|
||
|
||
hosts_data[host].append({
|
||
'date': r.file_date,
|
||
'os': r.os_name,
|
||
'ip': r.ip_address,
|
||
'compliance': r.compliance_percentage,
|
||
'passed': r.passed_count,
|
||
'failed': r.failed_count,
|
||
'total': r.total_checks,
|
||
'mandatory_percentage': r.mandatory_percentage,
|
||
'mandatory_passed': r.mandatory_passed,
|
||
'mandatory_total': r.mandatory_total,
|
||
'unit_name': user_file_map.get(r.filename, ''),
|
||
'uploaded_by': user_upload_map.get(r.filename, '')
|
||
})
|
||
|
||
# Đồng bộ đơn vị và tiêu chí bắt buộc giữa các lần quét của cùng 1 máy chủ trước khi xuất Excel
|
||
for host, points in hosts_data.items():
|
||
valid_unit = ""
|
||
valid_mp, valid_mt, valid_mpct = None, None, None
|
||
for pt in reversed(points):
|
||
if pt.get('unit_name') and pt['unit_name'] != '-':
|
||
valid_unit = pt['unit_name']
|
||
if pt.get('mandatory_total') and pt['mandatory_total'] > 0:
|
||
valid_mp = pt['mandatory_passed']
|
||
valid_mt = pt['mandatory_total']
|
||
valid_mpct = pt.get('mandatory_percentage')
|
||
if not valid_mpct or valid_mpct == 0:
|
||
valid_mpct = round((valid_mp / valid_mt) * 100, 1)
|
||
for pt in points:
|
||
if valid_unit and (not pt.get('unit_name') or pt['unit_name'] == '-'):
|
||
pt['unit_name'] = valid_unit
|
||
if valid_mt and (not pt.get('mandatory_total') or pt['mandatory_total'] == 0):
|
||
pt['mandatory_passed'] = valid_mp
|
||
pt['mandatory_total'] = valid_mt
|
||
pt['mandatory_percentage'] = valid_mpct if valid_mpct else round((valid_mp / valid_mt) * 100, 1)
|
||
if pt.get('mandatory_total', 0) > 0:
|
||
pt['mandatory_percentage'] = round((pt.get('mandatory_passed', 0) / pt['mandatory_total']) * 100, 1)
|
||
|
||
for host in hosts_data:
|
||
hosts_data[host].sort(key=lambda x: datetime.strptime(x['date'], '%d/%m/%Y %H:%M') if x['date'] else datetime.min)
|
||
|
||
# Build Excel
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.title = "Thong Ke May Chu"
|
||
|
||
# Styles
|
||
header_font = Font(name='Segoe UI', size=11, bold=True, color='FFFFFF')
|
||
header_fill = PatternFill(start_color='1565C0', end_color='1976D2', fill_type='solid')
|
||
header_align = Alignment(horizontal='center', vertical='center', wrap_text=True)
|
||
center_align = Alignment(horizontal='center', vertical='center')
|
||
thin_border = Border(
|
||
left=Side(style='thin'), right=Side(style='thin'),
|
||
top=Side(style='thin'), bottom=Side(style='thin')
|
||
)
|
||
|
||
headers = ["STT", "Tên Máy Chủ", "Đơn Vị", "Địa Chỉ IP", "Hệ Điều Hành", "Số Lần Quét", "Điểm Gần Nhất / Tổng", "Tỉ lệ Tuân thủ gần nhất (%)", "Tiêu chí bắt buộc", "Tiêu chí chưa đạt", "Người upload"]
|
||
for col, title in enumerate(headers, 1):
|
||
cell = ws.cell(row=1, column=col)
|
||
cell.value = title
|
||
cell.font = header_font
|
||
cell.fill = header_fill
|
||
cell.alignment = header_align
|
||
cell.border = thin_border
|
||
|
||
ws.column_dimensions['A'].width = 8
|
||
ws.column_dimensions['B'].width = 25
|
||
ws.column_dimensions['C'].width = 15
|
||
ws.column_dimensions['D'].width = 18
|
||
ws.column_dimensions['E'].width = 15
|
||
ws.column_dimensions['F'].width = 15
|
||
ws.column_dimensions['G'].width = 20
|
||
ws.column_dimensions['H'].width = 20
|
||
ws.column_dimensions['I'].width = 25
|
||
ws.column_dimensions['J'].width = 20
|
||
ws.column_dimensions['K'].width = 20
|
||
|
||
row_idx = 2
|
||
stt = 1
|
||
for host in hosts_data.keys():
|
||
data_points = hosts_data[host]
|
||
if not data_points:
|
||
continue
|
||
|
||
passes = [d['passed'] or 0 for d in data_points]
|
||
max_pass = max(passes) if passes else 0
|
||
latest = data_points[-1]
|
||
|
||
c = latest['compliance'] or 0.0
|
||
total_ck = latest['total'] or 0
|
||
|
||
status = f"{latest['mandatory_passed'] or 0}/{latest['mandatory_total'] or 0}"
|
||
|
||
row_data = [
|
||
stt,
|
||
host,
|
||
latest['unit_name'] or 'N/A',
|
||
latest['ip'] or 'N/A',
|
||
latest['os'] or 'N/A',
|
||
len(data_points),
|
||
f"{latest['passed']} / {total_ck}",
|
||
c,
|
||
status,
|
||
latest['failed'] or 0,
|
||
latest['uploaded_by'] or 'N/A'
|
||
]
|
||
|
||
for col_idx, val in enumerate(row_data, 1):
|
||
cell = ws.cell(row=row_idx, column=col_idx)
|
||
cell.value = val
|
||
cell.border = thin_border
|
||
if col_idx in [1, 5, 6, 7, 8, 9, 10]:
|
||
cell.alignment = center_align
|
||
row_idx += 1
|
||
stt += 1
|
||
|
||
output = io.BytesIO()
|
||
wb.save(output)
|
||
output.seek(0)
|
||
|
||
filename = f"ThongKe_MayChu_{datetime.now().strftime('%Y%m%d_%H%M%S')}.xlsx"
|
||
headers_response = {
|
||
'Content-Disposition': f'attachment; filename="{filename}"'
|
||
}
|
||
|
||
return Response(
|
||
content=output.getvalue(),
|
||
media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
|
||
headers=headers_response
|
||
)
|
||
|
||
finally:
|
||
db.close()
|
||
|
||
if __name__ == '__main__':
|
||
print(Fore.BLUE + r"""
|
||
_ _ _____ _____ _______ _ _ _____ _____ ______ _ _ _____ _ _ _____
|
||
/\ | | | | __ \_ _|__ __| | | | | /\ | __ \| __ \| ____| \ | |_ _| \ | |/ ____|
|
||
/ \ | | | | | | || | | | ______ | |__| | / \ | |__) | | | | |__ | \| | | | | \| | | __
|
||
/ /\ \| | | | | | || | | | |______| | __ | / /\ \ | _ /| | | | __| | . ` | | | | . ` | | |_ |
|
||
/ ____ \ |__| | |__| || |_ | | | | | |/ ____ \| | \ \| |__| | |____| |\ |_| |_| |\ | |__| |
|
||
/_/ \_\____/|_____/_____| |_| |_| |_/_/ \_\_| \_\_____/|______|_| \_|_____|_| \_\_____|
|
||
|
||
=== FASTAPI VERSION ===
|
||
""" + Fore.RESET)
|
||
print(Fore.GREEN + "\n>> Starting FastAPI application at http://127.0.0.1:8000" + Fore.RESET)
|
||
print(Fore.YELLOW + ">> Open your browser and go to: http://127.0.0.1:8000\n" + Fore.RESET)
|
||
print(Fore.CYAN + ">> API Docs available at: http://127.0.0.1:8000/docs" + Fore.RESET)
|
||
print(Fore.CYAN + ">> ReDoc available at: http://127.0.0.1:8000/redoc\n" + Fore.RESET)
|
||
|
||
uvicorn.run(
|
||
"web_app_fastapi:app",
|
||
host="0.0.0.0",
|
||
port=8000,
|
||
reload=True,
|
||
proxy_headers=True,
|
||
forwarded_allow_ips="*",
|
||
)
|