#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ ╔══════════════════════════════════════════════════════════════════════════════╗ ║ MAINSPRING PROJECT — ФИНАЛЬНАЯ БОЕВАЯ ВЕРСИЯ ║ ║ С ПРАВКАМИ ПО ПРАВИЛАМ #30, #42, #43, #46, #47, #32 ║ ║ + ПРОГРЕСС, ИЗБРАННОЕ, БАГРЕПОРТЫ, ЧАТ, АНАЛИТИКА ║ ║ Совместимость: Python 3.10+, Flask 2.3+, SQLite ║ ║ Дата: 12 августа 2026 | День разработки: 11 ║ ╚══════════════════════════════════════════════════════════════════════════════╝ """ import os import re import hashlib import hmac import secrets import logging import json import time import ipaddress import urllib.parse import random from datetime import datetime, timedelta from functools import wraps, lru_cache from collections import defaultdict from typing import Optional, Dict, Any, List, Tuple from html import escape from pathlib import Path from contextlib import contextmanager from flask import ( Flask, render_template, request, redirect, url_for, session, jsonify, abort, make_response, flash ) from werkzeug.security import generate_password_hash, check_password_hash from werkzeug.middleware.proxy_fix import ProxyFix import sqlite3 from config_loader import load_config # ============================================================================ # ИМПОРТ ИЗ db_utils.py — БЕЗ КОНФЛИКТОВ # ============================================================================ from db_utils import ( get_db, init_db, hash_password, log_action, register_user, check_user, is_banned, is_muted, find_anime_video ) # ============================================================================ # КОНФИГУРАЦИЯ # ============================================================================ class AppConfig: def __init__(self): raw = load_config() self.SECRET_KEY = raw.get('SECRET_KEY', secrets.token_hex(64)) self.DATABASE_PATH = os.environ.get('DATABASE_PATH', 'mainspring.db') self.ADMIN_LOGIN = 'admin' self.ADMIN_PASSWORD = raw.get('ADMIN_PASSWORD', secrets.token_urlsafe(12)) self.PASSWORD_MIN_LENGTH = 8 self.LOGIN_MIN_LENGTH = 3 self.SECRET_WORD_MIN_LENGTH = 3 self.LOGIN_PATTERN = re.compile(r'^[a-zA-Z0-9_\-]+$') self.MAX_LOGIN_ATTEMPTS = 5 self.LOGIN_BLOCK_MINUTES = 15 self.SESSION_LIFETIME_HOURS = 24 self.SESSION_COOKIE_SECURE = True self.SESSION_COOKIE_HTTPONLY = True self.SESSION_COOKIE_SAMESITE = 'Lax' self.DEFAULT_RATE_LIMIT = 100 self.DEFAULT_RATE_WINDOW = 60 self.API_RATE_LIMIT = 30 self.API_RATE_WINDOW = 60 self.COMMENT_RATE_LIMIT = 5 self.COMMENT_RATE_WINDOW = 300 self.DEFAULT_PER_PAGE = 15 self.SEED_TOKEN = raw.get('SEED_TOKEN', secrets.token_hex(32)) self.SEED_PASSWORD = raw.get('SEED_PASSWORD', secrets.token_hex(32)) self.KODIK_TOKEN = raw.get('KODIK_TOKEN', '') self.MAX_COMMENT_LENGTH = 500 self.MAX_USERNAME_LENGTH = 32 self.MAX_MESSAGE_LENGTH = 1000 self.CHAT_MESSAGES_PER_PAGE = 50 self.CAPTCHA_MIN = 1 self.CAPTCHA_MAX = 9 self.HONEYPOT_PATHS = [ '/wp-admin', '/wp-login.php', '/administrator', '/secret-admin', '/superadmin', '/.env', '/.git/config', '/phpmyadmin', '/adminer', '/sql', '/db', '/backup', '/old', '/test', '/dev', '/staging', '/api/v1', '/graphql', '/.aws/credentials', '/.ssh/id_rsa', '/config.php', '/wp-config.php', '/.htaccess', '/robots.txt', '/shell.php', '/cmd.php', '/upload.php', '/admin.php', '/wp-content/plugins', '/wp-content/themes', '/wp-includes', '/.well-known/security.txt', '/actuator', '/api/admin' ] self.POSTERS_DIR = Path('static/posters') self.AVATARS_DIR = Path('static/avatars') self.BACKUP_DB_DIR = Path('backups/db') self.LOG_DIR = Path('logs') def validate(self) -> List[str]: warnings = [] if len(self.SECRET_KEY) < 32: warnings.append("SECRET_KEY слишком короткий") if not self.KODIK_TOKEN: warnings.append("KODIK_TOKEN не настроен") if self.ADMIN_PASSWORD == 'admin': warnings.append("Пароль админа не изменён") return warnings config = AppConfig() for w in config.validate(): logging.warning(f"⚠️ {w}") # ============================================================================ # ЛОГИРОВАНИЕ # ============================================================================ config.LOG_DIR.mkdir(parents=True, exist_ok=True) console = logging.StreamHandler() console.setFormatter(logging.Formatter('%(asctime)s [%(levelname)s] %(name)s: %(message)s', datefmt='%H:%M:%S')) file_handler = logging.FileHandler(config.LOG_DIR / 'mainspring.log', encoding='utf-8') file_handler.setFormatter(logging.Formatter('%(asctime)s [%(levelname)s] %(name)s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S')) logger = logging.getLogger('mainspring') logger.setLevel(logging.INFO) logger.addHandler(console) logger.addHandler(file_handler) security_logger = logging.getLogger('mainspring.security') security_logger.setLevel(logging.WARNING) security_logger.addHandler(file_handler) # ============================================================================ # КЛАСС ДЛЯ ОШИБОК ПРИЛОЖЕНИЯ (Правило #43) # ============================================================================ class AppException(Exception): """Базовое исключение приложения с кодом и сообщением.""" def __init__(self, code: int, message: str, details: Optional[Dict] = None): self.code = code self.message = message self.details = details or {} super().__init__(message) class ValidationError(AppException): """Ошибка валидации (400).""" def __init__(self, message: str, details: Optional[Dict] = None): super().__init__(400, message, details) class AuthError(AppException): """Ошибка аутентификации (401/403).""" def __init__(self, message: str, code: int = 401, details: Optional[Dict] = None): super().__init__(code, message, details) class NotFoundError(AppException): """Ошибка 404.""" def __init__(self, message: str = "Не найдено"): super().__init__(404, message) # ============================================================================ # СЕРВИСЫ # ============================================================================ class PasswordService: HASH_METHOD = 'pbkdf2:sha256:600000' @classmethod def hash(cls, password: str) -> str: if not password or len(password) < config.PASSWORD_MIN_LENGTH: raise ValidationError(f"Пароль должен быть не менее {config.PASSWORD_MIN_LENGTH} символов") return generate_password_hash(password, method=cls.HASH_METHOD) @classmethod def verify(cls, password_hash: str, password: str) -> bool: if not password_hash or not password: return False try: return check_password_hash(password_hash, password) except Exception as e: logger.error(f"Ошибка проверки пароля: {e}") return False class RateLimiter: def __init__(self): self._storage = defaultdict(list) self._last_cleanup = datetime.now() self._cleanup_interval = 300 def is_allowed(self, key: str, limit: int = 100, window: int = 60) -> bool: self._maybe_cleanup() now = datetime.now() self._storage[key] = [t for t in self._storage[key] if (now - t).total_seconds() < window] if len(self._storage[key]) >= limit: return False self._storage[key].append(now) return True def get_remaining(self, key: str, limit: int = 100, window: int = 60) -> int: self._maybe_cleanup() now = datetime.now() self._storage[key] = [t for t in self._storage[key] if (now - t).total_seconds() < window] return max(0, limit - len(self._storage[key])) def get_retry_after(self, key: str, window: int = 60) -> int: if key not in self._storage or not self._storage[key]: return 0 oldest = min(self._storage[key]) return max(0, int(window - (datetime.now() - oldest).total_seconds())) def _maybe_cleanup(self): if (datetime.now() - self._last_cleanup).total_seconds() > self._cleanup_interval: self._cleanup() self._last_cleanup = datetime.now() def _cleanup(self): now = datetime.now() for key in list(self._storage.keys()): self._storage[key] = [t for t in self._storage[key] if (now - t).total_seconds() < 3600] if not self._storage[key]: del self._storage[key] class AuthService: _login_attempts: Dict[str, Dict] = {} @classmethod def check_bruteforce(cls, ip: str) -> Tuple[bool, Optional[str]]: if ip in cls._login_attempts: attempt = cls._login_attempts[ip] if attempt['count'] >= config.MAX_LOGIN_ATTEMPTS: if datetime.now() < attempt['blocked_until']: remaining = (attempt['blocked_until'] - datetime.now()).seconds // 60 return True, f"IP заблокирован на {remaining} мин." del cls._login_attempts[ip] return False, None @classmethod def register_attempt(cls, ip: str, success: bool): if success: cls._login_attempts.pop(ip, None) else: if ip not in cls._login_attempts: cls._login_attempts[ip] = {'count': 0, 'blocked_until': datetime.now()} cls._login_attempts[ip]['count'] += 1 if cls._login_attempts[ip]['count'] >= config.MAX_LOGIN_ATTEMPTS: cls._login_attempts[ip]['blocked_until'] = datetime.now() + timedelta(minutes=config.LOGIN_BLOCK_MINUTES) security_logger.warning(f"🔒 IP {ip} заблокирован") @classmethod def register_user(cls, login: str, password: str, secret_word: str, ip: str = None) -> Tuple[bool, str]: login = login.lower().strip() if len(login) < config.LOGIN_MIN_LENGTH: return False, f"Логин должен быть не менее {config.LOGIN_MIN_LENGTH} символов" if not config.LOGIN_PATTERN.match(login): return False, "Логин содержит недопустимые символы" return register_user(login, password, secret_word) @classmethod def authenticate(cls, login: str, password: str, ip: str = None) -> Tuple[bool, str, Optional[Dict]]: login = login.lower().strip() if ip: blocked, msg = cls.check_bruteforce(ip) if blocked: return False, msg, None try: if check_user(login, password): with get_db() as conn: c = conn.cursor() c.execute("SELECT id, login, role, avatar FROM users WHERE login = ?", (login,)) row = c.fetchone() if row: user_data = dict(row) log_action(login, 'Вход', ip) return True, "Успешный вход", { 'id': user_data['id'], 'login': user_data['login'], 'role': user_data['role'], 'avatar': user_data['avatar'] } except Exception as e: logger.error(f"Ошибка аутентификации: {e}") if ip: cls.register_attempt(ip, False) return False, "Внутренняя ошибка", None if ip: cls.register_attempt(ip, False) return False, "Неверный логин или пароль", None @classmethod def reset_password(cls, login: str, secret_word: str, new_password: str) -> Tuple[bool, str]: login = login.lower().strip() if len(new_password) < config.PASSWORD_MIN_LENGTH: return False, f"Пароль должен быть не менее {config.PASSWORD_MIN_LENGTH} символов" try: with get_db() as conn: c = conn.cursor() c.execute("SELECT secret_word FROM users WHERE login = ?", (login,)) row = c.fetchone() if not row: return False, "Пользователь не найден" if row['secret_word'] != secret_word: return False, "Неверное секретное слово" new_hash = PasswordService.hash(new_password) c.execute("UPDATE users SET password_hash = ? WHERE login = ?", (new_hash, login)) log_action(login, 'Сброс пароля') return True, f"Пароль изменён!" except Exception as e: logger.error(f"Ошибка сброса пароля: {e}") return False, "Внутренняя ошибка" class CSRFManager: TOKEN_LENGTH = 32 @classmethod def generate(cls) -> str: return secrets.token_hex(cls.TOKEN_LENGTH) @classmethod def validate(cls, form_token: Optional[str], session_token: Optional[str]) -> bool: if not form_token or not session_token: return False return hmac.compare_digest(form_token, session_token) @classmethod def rotate(cls): session['_csrf_token'] = cls.generate() session['_csrf_token_created'] = datetime.now().isoformat() class InputValidator: @staticmethod def sanitize_text(text: str, max_length: int = 500) -> str: if not text: return '' text = re.sub(r'[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]', '', text) text = text[:max_length] return escape(text).strip() @staticmethod def validate_username(username: str) -> Tuple[bool, str]: if not username: return False, "Имя не может быть пустым" username = username.strip() if len(username) > config.MAX_USERNAME_LENGTH: return False, f"Имя должно быть не более {config.MAX_USERNAME_LENGTH} символов" return True, InputValidator.sanitize_text(username, config.MAX_USERNAME_LENGTH) @staticmethod def validate_comment(comment: str) -> Tuple[bool, str]: if not comment or not comment.strip(): return False, "Комментарий не может быть пустым" comment = comment.strip() if len(comment) > config.MAX_COMMENT_LENGTH: return False, f"Комментарий должен быть не более {config.MAX_COMMENT_LENGTH} символов" if re.search(r' Tuple[bool, str]: if not url: return True, "" url = url.strip() parsed = urllib.parse.urlparse(url) if parsed.scheme not in ('http', 'https'): return False, "Недопустимая схема URL" dangerous = ['javascript:', 'data:', 'vbscript:', 'file:'] for p in dangerous: if p in url.lower(): return False, "URL содержит недопустимый протокол" return True, url[:2048] rate_limiter = RateLimiter() auth_service = AuthService() validator = InputValidator() # ============================================================================ # ВСПОМОГАТЕЛЬНЫЕ # ============================================================================ def record_visit(ip: str, user_agent: str = None) -> None: try: with get_db() as conn: conn.execute("INSERT INTO visits (ip, date, user_agent) VALUES (?, ?, ?)", (ip, datetime.now().strftime('%Y-%m-%d'), user_agent or '')) except Exception as e: logger.error(f"Ошибка записи посещения: {e}") def get_client_ip() -> str: if request.headers.get('X-Forwarded-For'): ips = request.headers['X-Forwarded-For'].split(',') try: ipaddress.ip_address(ips[0].strip()) return ips[0].strip() except ValueError: pass return request.remote_addr or '127.0.0.1' def generate_captcha() -> Tuple[int, str]: n1 = secrets.randbelow(config.CAPTCHA_MAX - config.CAPTCHA_MIN + 1) + config.CAPTCHA_MIN n2 = secrets.randbelow(config.CAPTCHA_MAX - config.CAPTCHA_MIN + 1) + config.CAPTCHA_MIN return n1 + n2, f'{n1} + {n2} = ?' def create_admin_if_not_exists() -> None: with get_db() as conn: c = conn.cursor() c.execute("SELECT id FROM users WHERE role = 'admin'") if not c.fetchone(): raw_password = secrets.token_urlsafe(12) password_hash = PasswordService.hash(raw_password) conn.execute( "INSERT INTO users (login, password_hash, avatar, secret_word, role) VALUES (?, ?, ?, ?, ?)", (config.ADMIN_LOGIN, password_hash, '', 'admin', 'admin') ) from db_crypto import xor_crypt, MASTER_KEY import base64 try: with open('config.enc', 'r') as f: encrypted = base64.b64decode(f.read()) decrypted = xor_crypt(encrypted, MASTER_KEY) cfg = json.loads(decrypted.decode()) except Exception as e: logger.warning(f"Не удалось загрузить config.enc для сохранения пароля: {e}") cfg = {} cfg['ADMIN_PASSWORD'] = raw_password json_str = json.dumps(cfg) encrypted = xor_crypt(json_str.encode(), MASTER_KEY) encoded = base64.b64encode(encrypted).decode() with open('config.enc', 'w') as f: f.write(encoded) logger.info("🔐 Администратор создан (пароль зашифрован)") print(f"\n🔐 АДМИН СОЗДАН:") print(f" Логин: {config.ADMIN_LOGIN}") print(f" Пароль: {raw_password}") print(" Пароль сохранён в config.enc\n") _stats_cache: Dict[str, Any] = {} _stats_cache_time: float = 0 def get_site_stats(force: bool = False) -> Dict[str, Any]: global _stats_cache, _stats_cache_time now = time.time() if not force and _stats_cache and (now - _stats_cache_time) < 60: return _stats_cache try: with get_db() as conn: c = conn.cursor() today = datetime.now().strftime('%Y-%m-%d') c.execute("SELECT COUNT(DISTINCT ip) FROM visits") total_visitors = c.fetchone()[0] or 0 c.execute("SELECT COUNT(DISTINCT ip) FROM visits WHERE date = ?", (today,)) today_visitors = c.fetchone()[0] or 0 c.execute("SELECT COUNT(*) FROM anime") total_anime = c.fetchone()[0] or 0 c.execute("SELECT COUNT(*) FROM manga") total_manga = c.fetchone()[0] or 0 c.execute("SELECT COUNT(*) FROM ranobe") total_ranobe = c.fetchone()[0] or 0 c.execute("SELECT COUNT(*) FROM anime_comments") total_comments = c.fetchone()[0] or 0 c.execute("SELECT COUNT(*) FROM users") total_users = c.fetchone()[0] or 0 _stats_cache = { 'total_visitors': total_visitors, 'today_visitors': today_visitors, 'total_anime': total_anime, 'total_manga': total_manga, 'total_ranobe': total_ranobe, 'total_comments': total_comments, 'total_users': total_users, 'online_users': 0, 'new_episodes_24h': 0, 'new_episodes_1h': 0, } _stats_cache_time = now return _stats_cache except Exception as e: logger.error(f"Ошибка сбора статистики: {e}") return _stats_cache or {'total_visitors': 0, 'today_visitors': 0, 'total_anime': 0, 'total_manga': 0, 'total_ranobe': 0, 'total_comments': 0, 'total_users': 0, 'online_users': 0, 'new_episodes_24h': 0, 'new_episodes_1h': 0} def get_anime_list_counts(anime_id: int) -> Dict[str, int]: """Возвращает количество пользователей по каждому статусу просмотра.""" try: with get_db() as conn: c = conn.cursor() c.execute("SELECT status, COUNT(*) as count FROM user_lists WHERE anime_id = ? GROUP BY status", (anime_id,)) return {row["status"]: row["count"] for row in c.fetchall()} except Exception: return {"watching": 0, "planned": 0, "completed": 0} # ============================================================================ # ДЕКОРАТОРЫ # ============================================================================ def login_required(f): @wraps(f) def decorated(*args, **kwargs): if 'user' not in session: return redirect(url_for('index')) try: banned, reason = is_banned(session['user']) if banned: session.clear() return render_template('index.html', message=f'Ваш аккаунт забанен: {reason}', captcha=session.get('captcha_question', '')) except Exception as e: logger.error(f"Ошибка проверки бана: {e}") return render_template('index.html', message='Внутренняя ошибка. Попробуйте позже.', captcha=session.get('captcha_question', '')) return f(*args, **kwargs) return decorated def admin_required(f): @wraps(f) def decorated(*args, **kwargs): if 'user' not in session or session.get('role') != 'admin': flash('Доступ запрещён.', 'error') return redirect(url_for('dashboard')) return f(*args, **kwargs) return decorated def csrf_protected(f): @wraps(f) def decorated(*args, **kwargs): if request.method == 'POST' and not request.is_json: form_token = request.form.get('csrf_token') session_token = session.get('_csrf_token') if not CSRFManager.validate(form_token, session_token): security_logger.warning(f"CSRF-нарушение: {request.path} | IP: {get_client_ip()}") return render_template('error.html', error_code=403, message='Ошибка безопасности. Обновите страницу и попробуйте снова.'), 403 CSRFManager.rotate() return f(*args, **kwargs) return decorated def api_rate_limit(limit: int = 30, window: int = 60): def decorator(f): @wraps(f) def decorated(*args, **kwargs): key = f"api:{get_client_ip()}:{request.path}" if not rate_limiter.is_allowed(key, limit, window): retry_after = rate_limiter.get_retry_after(key, window) return jsonify({'error': 'Too many requests', 'retry_after': retry_after}), 429 return f(*args, **kwargs) return decorated return decorator # ============================================================================ # FLASK APP # ============================================================================ app = Flask(__name__, static_folder='static') app.secret_key = config.SECRET_KEY app.config.update( SESSION_COOKIE_SECURE=config.SESSION_COOKIE_SECURE, SESSION_COOKIE_HTTPONLY=config.SESSION_COOKIE_HTTPONLY, SESSION_COOKIE_SAMESITE=config.SESSION_COOKIE_SAMESITE, PERMANENT_SESSION_LIFETIME=timedelta(hours=config.SESSION_LIFETIME_HOURS), MAX_CONTENT_LENGTH=16 * 1024 * 1024, ) app.wsgi_app = ProxyFix(app.wsgi_app, x_for=2, x_proto=1, x_host=1) def get_csrf(): if '_csrf_token' not in session: CSRFManager.rotate() return session['_csrf_token'] app.jinja_env.globals['csrf_token'] = get_csrf # ============================================================================ # SECURITY HEADERS # ============================================================================ @app.after_request def security_headers(response): response.headers['X-Content-Type-Options'] = 'nosniff' response.headers['X-Frame-Options'] = 'SAMEORIGIN' response.headers['X-XSS-Protection'] = '1; mode=block' response.headers['Strict-Transport-Security'] = 'max-age=31536000; includeSubDomains; preload' response.headers['Referrer-Policy'] = 'strict-origin-when-cross-origin' response.headers['Content-Security-Policy'] = ( "default-src 'self'; " "script-src 'self' 'unsafe-inline' 'unsafe-eval' https: blob:; " "style-src 'self' 'unsafe-inline' https:; " "img-src 'self' data: https:; " "font-src 'self' data: https:; " "frame-src 'self' https:; " "media-src 'self' https:; " "connect-src 'self' https: wss:; " "object-src 'none'; " "base-uri 'self'; " "form-action 'self'" ) response.headers['Permissions-Policy'] = 'camera=(), microphone=(), geolocation=()' if request.path.startswith('/static/'): response.headers['Cache-Control'] = 'public, max-age=31536000, immutable' return response # ============================================================================ # КОНТЕКСТ И ФИЛЬТРЫ # ============================================================================ @app.context_processor def inject_common(): return {'current_year': datetime.now().year, 'app_version': 'V199 Final', 'site_name': 'MainSpring'} @app.template_filter('poster_url') def poster_url_filter(item_id, image_url, item_type='anime'): local_path = f'static/posters/{item_type}_{item_id}.webp' if os.path.exists(local_path): return f'/static/posters/{item_type}_{item_id}.webp' if image_url and 'missing' not in str(image_url).lower(): return str(image_url) return '/static/img/default_poster.webp' @app.template_filter('time_ago') def time_ago_filter(date_str): if not date_str: return '' try: dt = datetime.fromisoformat(date_str.replace('Z', '+00:00')) if isinstance(date_str, str) else date_str now = datetime.now() diff = now - dt.replace(tzinfo=None) if diff.days > 365: return f"{diff.days // 365} г. назад" if diff.days > 30: return f"{diff.days // 30} мес. назад" if diff.days > 0: return f"{diff.days} д. назад" if diff.seconds > 3600: return f"{diff.seconds // 3600} ч. назад" if diff.seconds > 60: return f"{diff.seconds // 60} мин. назад" return "только что" except: return str(date_str) # ============================================================================ # ОБРАБОТЧИКИ ОШИБОК (Правило #46 — понятные сообщения) # ============================================================================ @app.errorhandler(404) def error_404(e): return render_template('error.html', error_code=404, message='Страница не найдена. Возможно, она была перемещена или удалена.'), 404 @app.errorhandler(500) def error_500(e): logger.error(f"Внутренняя ошибка: {e}", exc_info=True) return render_template('error.html', error_code=500, message='Что-то пошло не так. Мы уже работаем над исправлением.'), 500 @app.errorhandler(403) def error_403(e): return render_template('error.html', error_code=403, message='Доступ запрещён. У вас недостаточно прав.'), 403 @app.errorhandler(429) def error_429(e): return render_template('error.html', error_code=429, message='Слишком много запросов. Пожалуйста, подождите.'), 429 @app.errorhandler(400) def error_400(e): return render_template('error.html', error_code=400, message='Неверный запрос. Проверьте введённые данные.'), 400 # ============================================================================ # МАРШРУТЫ — ПУБЛИЧНЫЕ # ============================================================================ @app.route('/public') def public(): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT id, title_ru, rating, image_url FROM anime ORDER BY rating DESC LIMIT 10") top_anime = [dict(row) for row in c.fetchall()] return render_template('public_index.html', top_anime=top_anime) except Exception as e: logger.error(f"Ошибка на /public: {e}") return render_template('error.html', error_code=500, message='Не удалось загрузить главную страницу. Попробуйте позже.'), 500 @app.route('/', methods=['GET', 'POST']) @csrf_protected def index(): if 'user' in session: return redirect(url_for('dashboard')) ip = get_client_ip() record_visit(ip, request.headers.get('User-Agent')) if request.method == 'GET': session['captcha_result'], session['captcha_question'] = generate_captcha() session['comment_captcha_result'], session['comment_captcha_question'] = generate_captcha() message = '' if request.method == 'POST': try: blocked, msg = auth_service.check_bruteforce(ip) if blocked: return render_template('index.html', message=msg, captcha=session.get('captcha_question', '')) captcha_input = request.form.get('captcha_answer', '').strip() if not captcha_input.isdigit() or int(captcha_input) != session.get('captcha_result'): session['captcha_result'], session['captcha_question'] = generate_captcha() return render_template('index.html', message='Неверная капча!', captcha=session['captcha_question']) login = request.form.get('login', '').strip().lower() password = request.form.get('password', '') action = request.form.get('action') if not login or not password: message = 'Логин и пароль обязательны!' elif action == 'register': secret_word = request.form.get('secret_word', '').strip() if not secret_word: message = 'Придумайте секретное слово!' else: success, msg = auth_service.register_user(login, password, secret_word, ip) message = msg elif action == 'login': success, msg, user_data = auth_service.authenticate(login, password, ip) if success and user_data: session['user'] = user_data['login'] session['role'] = user_data['role'] session['avatar'] = user_data['avatar'] session.permanent = True return redirect(url_for('dashboard')) message = msg else: message = 'Неизвестное действие' except Exception as e: logger.error(f"Ошибка обработки POST на /: {e}") message = 'Внутренняя ошибка. Попробуйте позже.' return render_template('index.html', message=message, captcha=session.get('captcha_question', '')) @app.route('/forgot', methods=['GET', 'POST']) def forgot(): message = '' if request.method == 'POST': try: login = request.form.get('login', '').strip().lower() secret_word = request.form.get('secret_word', '').strip() new_password = request.form.get('new_password', '').strip() if not all([login, secret_word, new_password]): message = 'Все поля обязательны!' else: success, msg = auth_service.reset_password(login, secret_word, new_password) message = msg except Exception as e: logger.error(f"Ошибка восстановления пароля: {e}") message = 'Внутренняя ошибка. Попробуйте позже.' return render_template('forgot.html', message=message) # ============================================================================ # МАРШРУТЫ — ЗАЩИЩЁННЫЕ # ============================================================================ @app.route('/dashboard') @login_required def dashboard(): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT avatar, role FROM users WHERE login = ?", (session['user'],)) row = c.fetchone() avatar = row[0] if row and row[0] else '' role = row[1] if row and row[1] else 'user' stats = get_site_stats() return render_template('dashboard.html', user=session['user'], avatar=avatar, role=role, stats=stats) except Exception as e: logger.error(f"Ошибка на /dashboard: {e}") flash('Не удалось загрузить панель управления.', 'error') return redirect(url_for('public')) @app.route('/profile', methods=['GET', 'POST']) @login_required @csrf_protected def profile(): try: if request.method == 'POST': avatar = request.form.get('avatar', '').strip() is_valid, clean = validator.validate_url(avatar) with get_db() as conn: conn.execute("UPDATE users SET avatar = ? WHERE login = ?", (clean if is_valid else '', session['user'])) session['avatar'] = clean if is_valid else '' flash('Аватар обновлён!', 'success') with get_db() as conn: c = conn.cursor() c.execute("SELECT avatar FROM users WHERE login = ?", (session['user'],)) row = c.fetchone() return render_template('profile.html', user=session['user'], avatar=row[0] if row else '') except Exception as e: logger.error(f"Ошибка на /profile: {e}") flash('Ошибка обновления профиля.', 'error') return redirect(url_for('dashboard')) @app.route('/news') @login_required def news(): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT date, text FROM news_table ORDER BY id DESC LIMIT 50") news_list = [dict(row) for row in c.fetchall()] return render_template('news.html', user=session['user'], news=news_list) except Exception as e: logger.error(f"Ошибка на /news: {e}") flash('Ошибка загрузки новостей.', 'error') return redirect(url_for('dashboard')) @app.route('/download') @login_required def download(): return render_template('download.html', user=session['user']) @app.route('/rules') @login_required def rules(): return render_template('rules.html', user=session['user']) @app.route('/shop') @login_required def shop(): return render_template('shop.html', user=session['user'], items=[]) @app.route('/history') @login_required def history(): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT * FROM logs WHERE user = ? ORDER BY date DESC LIMIT 100", (session['user'],)) history_items = [dict(row) for row in c.fetchall()] return render_template('history.html', user=session['user'], history=history_items) except Exception as e: logger.error(f"Ошибка на /history: {e}") flash('Ошибка загрузки истории.', 'error') return redirect(url_for('dashboard')) @app.route('/servers') @login_required def servers(): return render_template('servers.html', user=session['user']) @app.route('/schedule') def schedule(): return render_template('schedule.html') @app.route('/arena') def arena(): return render_template('arena.html') # ============================================================================ # КАТАЛОГИ (с OFFSET Pagination) # ============================================================================ CATALOG_COLUMNS = "id, title_ru, title, rating, image_url, genres, year, episodes, slug" def _get_catalog_items(table: str, page: int = 1, genre: str = '', search: str = '', sort: str = 'rating', per_page: int = 15): valid_tables = ['anime', 'manga', 'ranobe'] valid_sorts = {'rating': 'rating DESC', 'year': 'year DESC', 'title': 'title_ru ASC'} if table not in valid_tables: return [], 0, 0, [], False, None sort_col = valid_sorts.get(sort, 'rating DESC') try: with get_db() as conn: c = conn.cursor() count_query = f"SELECT COUNT(*) FROM {table} WHERE 1=1" count_params = [] if genre: count_query += " AND genres LIKE ?" count_params.append(f'%{genre}%') if search: count_query += " AND (title_ru LIKE ? OR title LIKE ?)" count_params.extend([f'%{search}%', f'%{search}%']) c.execute(count_query, count_params) total = c.fetchone()[0] or 0 offset = (page - 1) * per_page query = f"SELECT {CATALOG_COLUMNS} FROM {table} WHERE 1=1" params = [] if genre: query += " AND genres LIKE ?" params.append(f'%{genre}%') if search: query += " AND (title_ru LIKE ? OR title LIKE ?)" params.extend([f'%{search}%', f'%{search}%']) query += f" ORDER BY {sort_col}, id ASC LIMIT ? OFFSET ?" params.extend([per_page, offset]) c.execute(query, params) rows = c.fetchall() items = [dict(row) for row in rows] total_pages = max(1, (total + per_page - 1) // per_page) c.execute(f"SELECT id, title_ru, rating, image_url FROM {table} ORDER BY rating DESC LIMIT 10") top = [dict(row) for row in c.fetchall()] return items, total, total_pages, top, False, None except Exception as e: logger.error(f"Ошибка получения каталога {table}: {e}") return [], 0, 0, [], False, None @app.route('/anime') def anime_list(): page = request.args.get('page', 1, type=int) items, total, pages, top, has_next, last_item = _get_catalog_items('anime', page, request.args.get('genre', ''), request.args.get('search', ''), request.args.get('sort', 'rating')) return render_template('anime_catalog.html', anime=items, page=page, total_pages=pages, total=total, top_anime=top, has_next=has_next, last_item=last_item) @app.route('/manga') def manga_list(): page = request.args.get('page', 1, type=int) items, total, pages, top, has_next, last_item = _get_catalog_items('manga', page, request.args.get('genre', ''), request.args.get('search', ''), request.args.get('sort', 'rating')) return render_template('manga_catalog.html', manga=items, page=page, total_pages=pages, total=total, top_manga=top, has_next=has_next, last_item=last_item) @app.route('/ranobe') def ranobe_list(): page = request.args.get('page', 1, type=int) items, total, pages, top, has_next, last_item = _get_catalog_items('ranobe', page, request.args.get('genre', ''), request.args.get('search', ''), request.args.get('sort', 'rating')) return render_template('ranobe_catalog.html', ranobe=items, page=page, total_pages=pages, total=total, top_ranobe=top, has_next=has_next, last_item=last_item) @app.route('/anime/') def anime_page(anime_id): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT * FROM anime WHERE id = ?", (anime_id,)) row = c.fetchone() if not row: return redirect(url_for('anime_list')) anime = dict(row) seasons = [{'id': 1, 'number': 1}] similar = [] if anime.get('genres'): first_genre = anime['genres'].split(',')[0].strip() with get_db() as conn: c = conn.cursor() c.execute("SELECT id, title_ru, rating, image_url FROM anime WHERE id != ? AND genres LIKE ? ORDER BY rating DESC LIMIT 6", (anime_id, f'%{first_genre}%')) similar = [dict(row) for row in c.fetchall()] list_counts = get_anime_list_counts(anime_id) return render_template('anime_page.html', anime=anime, similar=similar, seasons=seasons, list_counts=list_counts) except Exception as e: logger.error(f"Ошибка на /anime/{anime_id}: {e}") return render_template('error.html', error_code=500, message='Не удалось загрузить страницу аниме.'), 500 @app.route('/manga/') def manga_page(manga_id): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT * FROM manga WHERE id = ?", (manga_id,)) row = c.fetchone() if not row: return redirect(url_for('manga_list')) return render_template('manga_page.html', manga=dict(row)) except Exception as e: logger.error(f"Ошибка на /manga/{manga_id}: {e}") return render_template('error.html', error_code=500, message='Не удалось загрузить страницу манги.'), 500 @app.route('/ranobe/') def ranobe_page(ranobe_id): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT * FROM ranobe WHERE id = ?", (ranobe_id,)) row = c.fetchone() if not row: return redirect(url_for('ranobe_list')) return render_template('ranobe_page.html', ranobe=dict(row)) except Exception as e: logger.error(f"Ошибка на /ranobe/{ranobe_id}: {e}") return render_template('error.html', error_code=500, message='Не удалось загрузить страницу ранобэ.'), 500 @app.route('/random') def random_anime(): """Редиректит на случайное аниме из базы.""" try: with get_db() as conn: c = conn.cursor() c.execute("SELECT id FROM anime ORDER BY RANDOM() LIMIT 1") row = c.fetchone() if not row: abort(404) return redirect(url_for('anime_page', anime_id=row['id'])) except Exception as e: logger.error(f"Ошибка /random: {e}") abort(500) # ============================================================================ # API (Правило #32 — версионирование) # ============================================================================ @app.route('/api/v1/get_video/', methods=['GET']) @app.route('/api/get_video/', methods=['GET']) @api_rate_limit(30, 60) def api_get_video(anime_id): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT title, title_ru, slug FROM anime WHERE id = ?", (anime_id,)) row = c.fetchone() if not row: return jsonify({'error': 'Not found'}), 404 anime = dict(row) players = find_anime_video(anime.get('slug') or anime.get('title'), anime.get('slug')) return jsonify({'anime_id': anime_id, 'title': anime.get('title_ru') or anime.get('title'), 'players': players}) except Exception as e: logger.error(f"Ошибка API /get_video: {e}") return jsonify({'error': 'Внутренняя ошибка'}), 500 @app.route('/api/v1/comments/', methods=['GET']) @app.route('/api/comments/', methods=['GET']) @api_rate_limit(60, 60) def api_get_comments(item_id): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT username, comment, date FROM anime_comments WHERE anime_id = ? ORDER BY date DESC LIMIT 50", (item_id,)) comments = [dict(row) for row in c.fetchall()] return jsonify({'comments': comments}) except Exception as e: logger.error(f"Ошибка API /comments: {e}") return jsonify({'error': 'Внутренняя ошибка'}), 500 @app.route('/api/v1/add_comment/', methods=['POST']) @app.route('/api/add_comment/', methods=['POST']) @api_rate_limit(5, 300) def api_add_comment(item_id): try: data = request.get_json() if not data: return jsonify({'error': 'Нет данных'}), 400 captcha = str(data.get('captcha', '')).strip() if not captcha.isdigit() or int(captcha) != session.get('comment_captcha_result', -1): return jsonify({'error': 'Неверная капча'}), 403 username = data.get('username', 'Аноним') is_valid, username = validator.validate_username(username) if not is_valid: return jsonify({'error': 'Некорректное имя'}), 400 comment = data.get('comment', '') is_valid, comment = validator.validate_comment(comment) if not is_valid: return jsonify({'error': 'Некорректный комментарий'}), 400 with get_db() as conn: conn.execute("INSERT INTO anime_comments (anime_id, username, comment) VALUES (?, ?, ?)", (item_id, username, comment)) session['comment_captcha_result'], session['comment_captcha_question'] = generate_captcha() return jsonify({'status': 'ok'}) except Exception as e: logger.error(f"Ошибка API /add_comment: {e}") return jsonify({'error': 'Внутренняя ошибка'}), 500 @app.route('/api/v1/new_captcha', methods=['GET']) @app.route('/api/new_captcha', methods=['GET']) def api_new_captcha(): session['comment_captcha_result'], session['comment_captcha_question'] = generate_captcha() return jsonify({'question': session['comment_captcha_question']}) @app.route('/api/v1/news', methods=['GET']) @app.route('/api/news', methods=['GET']) @api_rate_limit(30, 60) def api_news(): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT date, text FROM news_table ORDER BY id DESC LIMIT 3") news = [dict(row) for row in c.fetchall()] return jsonify({'news': news}) except Exception as e: logger.error(f"Ошибка API /news: {e}") return jsonify({'error': 'Внутренняя ошибка'}), 500 @app.route('/api/v1/stats', methods=['GET']) @app.route('/api/stats', methods=['GET']) @login_required def api_stats(): return jsonify({'stats': {'play_time': '127 часов', 'mobs_killed': 3847, 'deaths': 23, 'coins': 15420}}) @app.route('/api/v1/auth_check', methods=['POST']) @app.route('/api/auth_check', methods=['POST']) def api_auth_check(): data = request.get_json() if data and data.get('token') == config.SEED_TOKEN and data.get('secret') == config.SEED_PASSWORD: return jsonify({'access': 'granted'}) return jsonify({'access': 'denied'}), 403 # ============================================================================ # НОВЫЕ API-МАРШРУТЫ (Прогресс, Избранное, Багрепорты, Чат, Аналитика) # ============================================================================ @app.route('/api/v1/progress/', methods=['POST']) @login_required def save_progress(anime_id: int): """Сохраняет прогресс просмотра (серия + таймкод).""" try: data = request.get_json(silent=True) if not data or 'episode' not in data or 'timestamp' not in data: return jsonify({"error": "Неверный формат данных"}), 400 with get_db() as conn: conn.execute( "INSERT OR REPLACE INTO user_progress (user_id, anime_id, episode, timestamp, updated_at) " "VALUES ((SELECT id FROM users WHERE login = ?), ?, ?, ?, datetime('now'))", (session['user'], anime_id, data['episode'], data['timestamp']) ) return jsonify({"status": "ok"}) except Exception as e: logger.error(f"Ошибка сохранения прогресса: {e}") return jsonify({"error": "Внутренняя ошибка"}), 500 @app.route('/api/v1/progress/', methods=['GET']) @login_required def get_progress(anime_id: int): """Возвращает сохранённый прогресс просмотра.""" try: with get_db() as conn: c = conn.cursor() c.execute( "SELECT episode, timestamp FROM user_progress WHERE user_id = (SELECT id FROM users WHERE login = ?) AND anime_id = ?", (session['user'], anime_id) ) row = c.fetchone() return jsonify(dict(row) if row else {"episode": 1, "timestamp": 0}) except Exception as e: logger.error(f"Ошибка получения прогресса: {e}") return jsonify({"error": "Внутренняя ошибка"}), 500 @app.route('/api/v1/favorite/', methods=['POST']) @login_required def toggle_favorite(anime_id: int): """Добавляет/удаляет аниме в избранное.""" try: with get_db() as conn: c = conn.cursor() c.execute("SELECT id FROM user_favorites WHERE user_id = (SELECT id FROM users WHERE login = ?) AND anime_id = ?", (session['user'], anime_id)) existing = c.fetchone() if existing: c.execute("DELETE FROM user_favorites WHERE id = ?", (existing['id'],)) return jsonify({"status": "removed"}) else: c.execute("INSERT INTO user_favorites (user_id, anime_id) VALUES ((SELECT id FROM users WHERE login = ?), ?)", (session['user'], anime_id)) return jsonify({"status": "added"}) except Exception as e: logger.error(f"Ошибка избранного: {e}") return jsonify({"error": "Внутренняя ошибка"}), 500 @app.route('/api/v1/report_bug/', methods=['POST']) @login_required def report_bug(anime_id: int): """Сохраняет багрепорт от пользователя.""" try: data = request.get_json(silent=True) if not data or 'message' not in data: return jsonify({"error": "Нет сообщения"}), 400 message = validator.sanitize_text(data['message'], 1000) with get_db() as conn: conn.execute( "INSERT INTO bug_reports (anime_id, user_id, message, created_at) " "VALUES (?, (SELECT id FROM users WHERE login = ?), ?, datetime('now'))", (anime_id, session['user'], message) ) logger.info(f"Багрепорт от {session['user']} на anime_id={anime_id}: {message}") return jsonify({"status": "ok"}) except Exception as e: logger.error(f"Ошибка багрепорта: {e}") return jsonify({"error": "Внутренняя ошибка"}), 500 @app.route('/api/v1/chat_messages', methods=['GET']) def get_community_chat(): """Возвращает последние 30 сообщений общего чата.""" try: with get_db() as conn: c = conn.cursor() c.execute( "SELECT u.login as user, cm.text, cm.created_at FROM community_chat cm " "JOIN users u ON cm.user_id = u.id ORDER BY cm.created_at DESC LIMIT 30" ) messages = [dict(row) for row in c.fetchall()] return jsonify({"messages": list(reversed(messages))}) except Exception as e: logger.error(f"Ошибка чата сообщества: {e}") return jsonify({"error": "Внутренняя ошибка"}), 500 @app.route('/api/v1/chat_send', methods=['POST']) @login_required def send_community_message(): """Отправляет сообщение в общий чат.""" try: data = request.get_json(silent=True) if not data or 'text' not in data: return jsonify({"error": "Пустое сообщение"}), 400 text = validator.sanitize_text(data['text'], config.MAX_MESSAGE_LENGTH) if not text: return jsonify({"error": "Сообщение не может быть пустым"}), 400 with get_db() as conn: conn.execute( "INSERT INTO community_chat (user_id, text, created_at) VALUES ((SELECT id FROM users WHERE login = ?), ?, datetime('now'))", (session['user'], text) ) return jsonify({"status": "ok"}) except Exception as e: logger.error(f"Ошибка отправки сообщения в чат: {e}") return jsonify({"error": "Внутренняя ошибка"}), 500 @app.route('/api/v1/track_click', methods=['POST']) def track_click(): """Логирует клик по элементу с data-track-click.""" data = request.get_json(silent=True) if data: logger.info(f"CLICK | page={data.get('page')} | element={data.get('element_id')} | ip={get_client_ip()}") return jsonify({"status": "ok"}) @app.route('/api/v1/engaged_visit', methods=['POST']) def engaged_visit(): """Фиксирует вовлечённый визит (более 15 секунд).""" data = request.get_json(silent=True) if data: logger.info(f"ENGAGED | page={data.get('page')} | time={data.get('time_spent')}s | ip={get_client_ip()}") return jsonify({"status": "ok"}) @app.route('/api/v1/set_anime_status', methods=['POST']) @login_required def set_anime_status(): """Устанавливает статус аниме для пользователя (watching/planned/completed).""" try: data = request.get_json(silent=True) if not data or 'anime_id' not in data or 'status' not in data: return jsonify({"error": "Неверный формат"}), 400 valid_statuses = ('watching', 'planned', 'completed') if data['status'] not in valid_statuses: return jsonify({"error": "Недопустимый статус"}), 400 with get_db() as conn: conn.execute( "INSERT OR REPLACE INTO user_lists (user_id, anime_id, status, updated_at) " "VALUES ((SELECT id FROM users WHERE login = ?), ?, ?, datetime('now'))", (session['user'], data['anime_id'], data['status']) ) return jsonify({"status": "ok"}) except Exception as e: logger.error(f"Ошибка установки статуса: {e}") return jsonify({"error": "Внутренняя ошибка"}), 500 # ============================================================================ # АДМИН # ============================================================================ @app.route('/admin') @admin_required def admin(): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT login, role, created_at FROM users ORDER BY created_at DESC") users = [dict(row) for row in c.fetchall()] stats = get_site_stats() return render_template('admin.html', user=session['user'], users=users, stats=stats) except Exception as e: logger.error(f"Ошибка на /admin: {e}") flash('Ошибка загрузки админ-панели.', 'error') return redirect(url_for('dashboard')) @app.route('/admin/logs') @admin_required def admin_logs(): try: page = request.args.get('page', 1, type=int) per_page = 50 with get_db() as conn: c = conn.cursor() c.execute("SELECT COUNT(*) FROM logs") total = c.fetchone()[0] c.execute("SELECT * FROM logs ORDER BY id DESC LIMIT ? OFFSET ?", (per_page, (page - 1) * per_page)) logs = [dict(row) for row in c.fetchall()] return render_template('logs.html', user=session['user'], logs=logs, page=page, total_pages=max(1, (total + per_page - 1) // per_page)) except Exception as e: logger.error(f"Ошибка на /admin/logs: {e}") flash('Ошибка загрузки логов.', 'error') return redirect(url_for('admin')) @app.route('/admin/add_news', methods=['POST']) @admin_required @csrf_protected def admin_add_news(): try: text = request.form.get('text', '').strip() if text: with get_db() as conn: conn.execute("INSERT INTO news_table (date, text) VALUES (?, ?)", (datetime.now().strftime('%Y-%m-%d %H:%M'), text)) flash('Новость добавлена!', 'success') return redirect(url_for('admin')) except Exception as e: logger.error(f"Ошибка добавления новости: {e}") flash('Ошибка добавления новости.', 'error') return redirect(url_for('admin')) @app.route('/admin/delete_news/') @admin_required def admin_delete_news(news_id): try: with get_db() as conn: conn.execute("DELETE FROM news_table WHERE id = ?", (news_id,)) flash('Новость удалена.', 'info') return redirect(url_for('admin')) except Exception as e: logger.error(f"Ошибка удаления новости: {e}") flash('Ошибка удаления новости.', 'error') return redirect(url_for('admin')) @app.route('/admin/delete_user/') @admin_required def admin_delete_user(login): if login == config.ADMIN_LOGIN: flash('Нельзя удалить администратора!', 'error') return redirect(url_for('admin')) try: with get_db() as conn: conn.execute("DELETE FROM users WHERE login = ?", (login,)) flash(f'Пользователь {login} удалён.', 'success') return redirect(url_for('admin')) except Exception as e: logger.error(f"Ошибка удаления пользователя: {e}") flash('Ошибка удаления пользователя.', 'error') return redirect(url_for('admin')) @app.route('/admin/ban_user/', methods=['POST']) @admin_required @csrf_protected def admin_ban_user(login): if login == config.ADMIN_LOGIN: flash('Нельзя забанить администратора!', 'error') return redirect(url_for('admin')) try: duration_days = int(request.form.get('duration', '7')) reason = request.form.get('reason', 'Нарушение правил').strip() banned_until = datetime.now() + timedelta(days=duration_days) with get_db() as conn: conn.execute("INSERT INTO bans (user, reason, banned_until, banned_by) VALUES (?, ?, ?, ?)", (login, reason, banned_until.isoformat(), session['user'])) flash(f'Пользователь {login} забанен на {duration_days} дн.', 'success') return redirect(url_for('admin')) except Exception as e: logger.error(f"Ошибка бана пользователя: {e}") flash('Ошибка бана пользователя.', 'error') return redirect(url_for('admin')) @app.route('/admin/unban_user/') @admin_required def admin_unban_user(login): try: with get_db() as conn: conn.execute("DELETE FROM bans WHERE user = ? AND banned_until > ?", (login, datetime.now().isoformat())) flash(f'Пользователь {login} разбанен.', 'success') return redirect(url_for('admin')) except Exception as e: logger.error(f"Ошибка разбана пользователя: {e}") flash('Ошибка разбана пользователя.', 'error') return redirect(url_for('admin')) # ============================================================================ # ЧАТ # ============================================================================ @app.route('/chat') @login_required def chat(): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT login, avatar FROM users WHERE login != ? ORDER BY login", (session['user'],)) users = [dict(row) for row in c.fetchall()] return render_template('chat.html', user=session['user'], users=users) except Exception as e: logger.error(f"Ошибка на /chat: {e}") flash('Ошибка загрузки чата.', 'error') return redirect(url_for('dashboard')) @app.route('/chat/', methods=['GET', 'POST']) @login_required @csrf_protected def chat_with(receiver): try: muted, mute_msg = is_muted(session['user']) if muted and request.method == 'POST': flash(f'Вы замучены: {mute_msg}', 'error') return redirect(url_for('chat')) if request.method == 'POST': text = request.form.get('text', '').strip() if text and len(text) <= config.MAX_MESSAGE_LENGTH: with get_db() as conn: conn.execute("INSERT INTO messages (sender, receiver, text) VALUES (?, ?, ?)", (session['user'], receiver, validator.sanitize_text(text, config.MAX_MESSAGE_LENGTH))) with get_db() as conn: c = conn.cursor() c.execute("""SELECT sender, receiver, text, date FROM messages WHERE (sender=? AND receiver=?) OR (sender=? AND receiver=?) ORDER BY date DESC LIMIT ?""", (session['user'], receiver, receiver, session['user'], config.CHAT_MESSAGES_PER_PAGE)) messages = [dict(row) for row in c.fetchall()] c.execute("SELECT login, avatar FROM users WHERE login != ? ORDER BY login", (session['user'],)) users = [dict(row) for row in c.fetchall()] return render_template('chat.html', user=session['user'], users=users, receiver=receiver, messages=messages) except Exception as e: logger.error(f"Ошибка в чате с {receiver}: {e}") flash('Ошибка загрузки чата.', 'error') return redirect(url_for('chat')) # ============================================================================ # P2P # ============================================================================ p2p_players: Dict[str, Dict] = {} @app.route('/p2p/register', methods=['POST']) @login_required def p2p_register(): try: data = request.get_json() if not data: return jsonify({'error': 'No data'}), 400 port = data.get('port', 25565) if not isinstance(port, int) or port < 1024 or port > 65535: return jsonify({'error': 'Invalid port'}), 400 p2p_players[session['user']] = { 'ip_hash': hashlib.sha256(get_client_ip().encode()).hexdigest()[:16], 'port': port, 'last_seen': datetime.now().isoformat(), 'version': data.get('version', '1.20.1') } return jsonify({'status': 'ok'}) except Exception as e: logger.error(f"Ошибка P2P регистрации: {e}") return jsonify({'error': 'Внутренняя ошибка'}), 500 @app.route('/p2p/players') @login_required def p2p_players_list(): try: now = datetime.now() active = {} for player, pdata in p2p_players.items(): if (now - datetime.fromisoformat(pdata['last_seen'])).total_seconds() < 300: active[player] = {'port': pdata['port'], 'version': pdata['version']} return jsonify({'players': active}) except Exception as e: logger.error(f"Ошибка получения списка P2P: {e}") return jsonify({'error': 'Внутренняя ошибка'}), 500 # ============================================================================ # HONEYPOT (Правило #30 — логирование в БД) # ============================================================================ FAKE_FILES = { '.env': ( "# Production Environment Configuration\n" "# Generated: " + datetime.now().strftime('%Y-%m-%d %H:%M:%S') + "\n" "# Server: mainspring-mc.ru\n\n" "APP_ENV=production\n" "APP_DEBUG=false\n" "APP_URL=https://mainspring-mc.ru\n\n" "DB_CONNECTION=sqlite\n" "DB_DATABASE=/var/www/mainspring/database/mainspring.db\n\n" "SECRET_KEY=prod_secret_key_x7k9m2w4n8p3q5r6t1v8y0a3b6c9d2e5\n" "ADMIN_TOKEN=admin_x7k9m2w4n8p3q5r6t1v8y0a3b6c9d2e5\n\n" "REDIS_HOST=localhost\n" "REDIS_PASSWORD=redis_prod_pass_123\n" "REDIS_PORT=6379\n\n" "MAIL_DRIVER=smtp\n" "MAIL_HOST=smtp.gmail.com\n" "MAIL_PORT=587\n" "MAIL_USERNAME=noreply@mainspring-mc.ru\n" "MAIL_PASSWORD=email_pass_12345\n\n" "# " + "x" * 400 + "\n# End of file\n" ), '.env.local': ( "# Local Development\n" "APP_ENV=local\n" "APP_DEBUG=true\n" "DB_HOST=localhost\n" "DB_USER=root\n" "DB_PASS=root123\n" "# " + "x" * 500 + "\n" ), '.git/config': ( "[core]\n" " repositoryformatversion = 0\n" " filemode = true\n" " bare = false\n" " logallrefupdates = true\n" "[remote \"origin\"]\n" " url = git@github.com:mainspring/mc-server.git\n" " fetch = +refs/heads/*:refs/remotes/origin/*\n" "[branch \"main\"]\n" " remote = origin\n" " merge = refs/heads/main\n" ), 'config.php': ( "" ), '.aws/credentials': ( "[default]\n" "aws_access_key_id = AKIAIOSFODNN7EXAMPLE\n" "aws_secret_access_key = wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY\n" "region = us-east-1\n" "output = json\n" ), '.ssh/id_rsa': ( "-----BEGIN RSA PRIVATE KEY-----\n" "MIIEowIBAAKCAQEAuYq9xK7mN3pQr5tV1wX8yA0bC3dE6fG9hI2jK5lM8nO1pQ4r\n" "S7tU9vW0xY1zA2B3cD4eF5gH6iJ7kL8mN9oP0qR1sT2uV3wX4yZ5a6b7c8d9e0f1\n" "g2h3i4j5k6l7m8n9o0p1q2r3s4t5u6v7w8x9y0z1a2b3c4d5e6f7g8h9i0j1k2l3m4\n" "n5o6p7q8r9s0t1u2v3w4x5y6z7a8b9c0d1e2f3g4h5i6j7k8l9m0n1o2p3q4r5s6t7\n" "u8v9w0x1y2z3a4b5c6d7e8f9g0h1i2j3k4l5m6n7o8p9q0r1s2t3u4v5w6x7y8z9\n" "-----END RSA PRIVATE KEY-----\n" ), } HONEYPOT_SET = set(config.HONEYPOT_PATHS) HONEYPOT_PREFIXES = ('/wp-', '/.', '/adminer', '/phpmyadmin', '/shell', '/cmd', '/config.', '/sql', '/db', '/backup') @app.route('/') def catch_all(path): full_path = f'/{path}' if full_path in HONEYPOT_SET or any(full_path.startswith(p) for p in HONEYPOT_PREFIXES): log_action('HONEYPOT', f'Попытка доступа к {full_path}', get_client_ip()) security_logger.warning(f"🍯 Honeypot: IP={get_client_ip()} Path={full_path}") filename = path.split('/')[-1] if filename in FAKE_FILES: response = make_response(FAKE_FILES[filename]) response.headers['Content-Type'] = 'text/plain' response.headers['Content-Disposition'] = f'attachment; filename="{filename}"' return response return render_template('hacked.html', visitor_ip=get_client_ip(), visit_time=datetime.now().strftime('%Y-%m-%d %H:%M:%S'), targeted_path=full_path) abort(404) # ============================================================================ # HEALTHCHECK (Правило #21, #33) # ============================================================================ @app.route('/health') def health(): try: with get_db() as conn: c = conn.cursor() c.execute("SELECT 1") return jsonify({'status': 'ok', 'database': 'connected', 'timestamp': datetime.now().isoformat()}), 200 except Exception as e: logger.error(f"Healthcheck failed: {e}") return jsonify({'status': 'error', 'database': str(e), 'timestamp': datetime.now().isoformat()}), 500 # ============================================================================ # ВЫХОД И ЗАПУСК # ============================================================================ @app.route('/logout') def logout(): if 'user' in session: log_action(session['user'], 'Выход', get_client_ip()) session.clear() return redirect(url_for('index')) def bootstrap(): logger.info("=" * 60) logger.info("🚀 MainSpring — ФИНАЛЬНАЯ БОЕВАЯ ВЕРСИЯ (С ПРАВКАМИ)") logger.info("=" * 60) config.BACKUP_DB_DIR.mkdir(parents=True, exist_ok=True) config.POSTERS_DIR.mkdir(parents=True, exist_ok=True) config.AVATARS_DIR.mkdir(parents=True, exist_ok=True) init_db() create_admin_if_not_exists() with get_db() as conn: c = conn.cursor() c.execute("SELECT COUNT(*) FROM anime") anime_count = c.fetchone()[0] or 0 c.execute("SELECT COUNT(*) FROM manga") manga_count = c.fetchone()[0] or 0 c.execute("SELECT COUNT(*) FROM ranobe") ranobe_count = c.fetchone()[0] or 0 logger.info(f"📊 Контент: {anime_count} аниме, {manga_count} манги, {ranobe_count} ранобэ") logger.info("=" * 60) logger.info("✅ Приложение готово к работе") logger.info("=" * 60) with app.app_context(): bootstrap() if __name__ == '__main__': app.run(host='0.0.0.0', port=int(os.environ.get('PORT', 5000)), debug=False, threaded=True)