InviteMe Developer
331 lines
14 KiB
Python
331 lines
14 KiB
Python
import os
|
|
import logging
|
|
import secrets
|
|
from pathlib import Path
|
|
|
|
import click
|
|
from flask import Flask, jsonify, redirect, request
|
|
from flask_sqlalchemy import SQLAlchemy
|
|
from flask_limiter import Limiter
|
|
from flask_limiter.util import get_remote_address
|
|
from flask_migrate import Migrate
|
|
from flask_wtf.csrf import CSRFProtect
|
|
from sqlalchemy import inspect, text
|
|
from werkzeug.middleware.proxy_fix import ProxyFix
|
|
|
|
from config import Config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
db = SQLAlchemy()
|
|
limiter = Limiter(key_func=get_remote_address)
|
|
csrf = CSRFProtect()
|
|
migrate = Migrate()
|
|
|
|
|
|
def create_app(config_class=Config):
|
|
if hasattr(config_class, "validate"):
|
|
config_class.validate()
|
|
|
|
app = Flask(__name__)
|
|
app.config.from_object(config_class)
|
|
|
|
if app.config.get("TRUST_PROXY_HEADERS"):
|
|
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1)
|
|
|
|
# Ensure upload directories exist
|
|
os.makedirs(app.config["PHOTO_FOLDER"], exist_ok=True)
|
|
os.makedirs(app.config["QR_FOLDER"], exist_ok=True)
|
|
os.makedirs(app.config["LOGO_FOLDER"], exist_ok=True)
|
|
database_uri = app.config["SQLALCHEMY_DATABASE_URI"]
|
|
if database_uri.startswith("sqlite:///"):
|
|
os.makedirs(os.path.dirname(database_uri.replace("sqlite:///", "", 1)), exist_ok=True)
|
|
|
|
db.init_app(app)
|
|
limiter.init_app(app)
|
|
csrf.init_app(app)
|
|
migrate.init_app(app, db)
|
|
|
|
from app.routes import main_bp
|
|
app.register_blueprint(main_bp)
|
|
_register_security_hooks(app)
|
|
_register_cli(app)
|
|
|
|
with app.app_context():
|
|
if not app.config.get("SKIP_DB_BOOTSTRAP"):
|
|
_run_migrations()
|
|
if not app.config.get("IS_PRODUCTION"):
|
|
_migrate_existing_data(app)
|
|
_migrate_legacy_public_uploads(app)
|
|
_load_smtp_settings(app)
|
|
|
|
return app
|
|
|
|
|
|
def _run_migrations():
|
|
"""Применяет миграции Alembic к текущей базе данных."""
|
|
from alembic.config import Config as AlembicConfig
|
|
from alembic import command
|
|
|
|
migrations_dir = Path(__file__).resolve().parent.parent / "migrations"
|
|
alembic_ini = migrations_dir / "alembic.ini"
|
|
alembic_cfg = AlembicConfig(str(alembic_ini))
|
|
alembic_cfg.set_main_option("script_location", str(migrations_dir))
|
|
command.upgrade(alembic_cfg, "head")
|
|
|
|
|
|
def _load_smtp_settings(app):
|
|
"""Загружает SMTP-настройки из базы в конфиг приложения."""
|
|
try:
|
|
from app.models import SmtpSettings
|
|
settings = SmtpSettings.query.first()
|
|
if settings and settings.encrypt_legacy_passwords():
|
|
db.session.commit()
|
|
SmtpSettings.apply_to_app(app)
|
|
except Exception as e:
|
|
logger.warning(f"SMTP settings load skipped: {e}")
|
|
|
|
|
|
def _column_exists(table_name, column_name):
|
|
inspector = inspect(db.engine)
|
|
if table_name not in inspector.get_table_names():
|
|
return False
|
|
return any(column["name"] == column_name for column in inspector.get_columns(table_name))
|
|
|
|
|
|
def _add_column_if_missing(table_name, column_name, column_def):
|
|
if not _column_exists(table_name, column_name):
|
|
db.session.execute(text(f"ALTER TABLE {table_name} ADD COLUMN {column_name} {column_def}"))
|
|
|
|
|
|
def _safe_move(src, dst):
|
|
"""Перемещает файл, игнорируя ошибки."""
|
|
try:
|
|
import shutil
|
|
dst.parent.mkdir(parents=True, exist_ok=True)
|
|
shutil.move(str(src), str(dst))
|
|
return True
|
|
except Exception as e:
|
|
logger.warning(f"Не удалось переместить {src} -> {dst}: {e}")
|
|
return False
|
|
|
|
|
|
def _migrate_legacy_public_uploads(app):
|
|
"""Move files from old public static/uploads into the private data directory."""
|
|
legacy_root = Path(app.root_path) / "static" / "uploads"
|
|
private_root = Path(app.config["UPLOAD_FOLDER"])
|
|
if not legacy_root.exists() or legacy_root.resolve() == private_root.resolve():
|
|
return
|
|
|
|
for source in legacy_root.rglob("*"):
|
|
if not source.is_file() or source.name == ".gitkeep":
|
|
continue
|
|
destination = private_root / source.relative_to(legacy_root)
|
|
if destination.exists():
|
|
destination = private_root / "legacy_conflicts" / source.relative_to(legacy_root)
|
|
if destination.exists():
|
|
destination = destination.with_name(f"{destination.stem}-{secrets.token_hex(4)}{destination.suffix}")
|
|
logger.warning("Legacy upload collision; preserving old file at: %s", destination)
|
|
_safe_move(source, destination)
|
|
|
|
|
|
def _register_security_hooks(app):
|
|
@app.before_request
|
|
def enforce_https():
|
|
if request.path != "/healthz" and app.config.get("ENFORCE_HTTPS") and not request.is_secure:
|
|
return redirect(request.url.replace("http://", "https://", 1), code=308)
|
|
|
|
@app.after_request
|
|
def add_security_headers(response):
|
|
response.headers.setdefault("X-Content-Type-Options", "nosniff")
|
|
response.headers.setdefault("X-Frame-Options", "DENY")
|
|
response.headers.setdefault("Referrer-Policy", "no-referrer")
|
|
response.headers.setdefault("Permissions-Policy", "camera=(self), microphone=(), geolocation=()")
|
|
response.headers.setdefault(
|
|
"Content-Security-Policy",
|
|
"default-src 'self'; "
|
|
"base-uri 'self'; frame-ancestors 'none'; form-action 'self'; object-src 'none'; "
|
|
"img-src 'self' data:; connect-src 'self'; "
|
|
"style-src 'self' 'unsafe-inline' https://cdn.jsdelivr.net; "
|
|
"script-src 'self' https://cdn.jsdelivr.net https://unpkg.com",
|
|
)
|
|
if request.is_secure:
|
|
response.headers.setdefault("Strict-Transport-Security", "max-age=31536000; includeSubDomains")
|
|
if request.path.startswith(("/admin", "/api", "/check-in", "/media", "/invite")):
|
|
response.headers.setdefault("Cache-Control", "private, no-store")
|
|
return response
|
|
|
|
@app.errorhandler(413)
|
|
def request_too_large(_error):
|
|
return jsonify({"error": "Загружаемый файл слишком большой"}), 413
|
|
|
|
@app.get("/healthz")
|
|
@limiter.exempt
|
|
def healthz():
|
|
db.session.execute(text("SELECT 1"))
|
|
return jsonify({"status": "ok"})
|
|
|
|
|
|
def _register_cli(app):
|
|
@app.cli.command("create-admin")
|
|
@click.option("--username", prompt=True)
|
|
@click.password_option(confirmation_prompt=True)
|
|
def create_admin(username, password):
|
|
"""Create the first administrator without exposing a web setup page."""
|
|
from app.models import User
|
|
from app.utils.security import validate_password
|
|
|
|
username = username.strip()
|
|
error = validate_password(password, app.config["MIN_PASSWORD_LENGTH"])
|
|
if not username:
|
|
raise click.ClickException("Логин не может быть пустым")
|
|
if error:
|
|
raise click.ClickException(error)
|
|
if User.query.filter_by(username=username).first():
|
|
raise click.ClickException("Пользователь уже существует")
|
|
|
|
user = User(username=username, role="admin", is_active=True)
|
|
user.set_password(password)
|
|
db.session.add(user)
|
|
db.session.commit()
|
|
click.echo("Администратор создан")
|
|
|
|
|
|
def _migrate_files_to_event_folders():
|
|
"""Переносит существующие файлы в папки по мероприятиям."""
|
|
from pathlib import Path
|
|
from flask import current_app
|
|
from app.models import Event, Guest
|
|
from app.utils.paths import safe_child_path
|
|
|
|
photo_base = Path(current_app.config["PHOTO_FOLDER"])
|
|
qr_base = Path(current_app.config["QR_FOLDER"])
|
|
logo_base = Path(current_app.config["LOGO_FOLDER"])
|
|
tickets_base = Path(current_app.config["UPLOAD_FOLDER"]) / "tickets"
|
|
|
|
# Фото гостей
|
|
for guest in Guest.query.filter(Guest.photo_filename.isnot(None)).all():
|
|
if not guest.photo_filename:
|
|
continue
|
|
src = safe_child_path(photo_base, guest.photo_filename)
|
|
if not src:
|
|
continue
|
|
if src.exists() and src.is_file():
|
|
dst = safe_child_path(photo_base / f"event_{guest.event_id}", guest.photo_filename)
|
|
_safe_move(src, dst)
|
|
|
|
# QR-коды гостей
|
|
for guest in Guest.query.filter(Guest.qr_filename.isnot(None)).all():
|
|
if not guest.qr_filename:
|
|
continue
|
|
src = safe_child_path(qr_base, guest.qr_filename)
|
|
if not src:
|
|
continue
|
|
if src.exists() and src.is_file():
|
|
dst = safe_child_path(qr_base / f"event_{guest.event_id}", guest.qr_filename)
|
|
_safe_move(src, dst)
|
|
|
|
# Логотипы мероприятий
|
|
for event in Event.query.filter(Event.logo_filename.isnot(None)).all():
|
|
if not event.logo_filename:
|
|
continue
|
|
src = safe_child_path(logo_base, event.logo_filename)
|
|
if not src:
|
|
continue
|
|
if src.exists() and src.is_file():
|
|
dst = safe_child_path(logo_base / f"event_{event.id}", event.logo_filename)
|
|
_safe_move(src, dst)
|
|
|
|
# Сгенерированные билеты
|
|
if tickets_base.exists():
|
|
for event in Event.query.all():
|
|
event_tickets_dir = tickets_base / f"event_{event.id}"
|
|
event_tickets_dir.mkdir(parents=True, exist_ok=True)
|
|
for ticket_file in tickets_base.glob(f"ticket_{event.id}_*.png"):
|
|
_safe_move(ticket_file, event_tickets_dir / ticket_file.name)
|
|
|
|
|
|
def _migrate_existing_data(app):
|
|
"""Миграция существующих гостей в дефолтное мероприятие и добавление новых колонок."""
|
|
from app.models import Event, Guest
|
|
|
|
try:
|
|
# Добавляем колонки дизайна билета в таблицу events
|
|
_add_column_if_missing("events", "ticket_title", "VARCHAR(255)")
|
|
_add_column_if_missing("events", "ticket_subtitle", "VARCHAR(255)")
|
|
_add_column_if_missing("events", "ticket_body_text", "TEXT")
|
|
_add_column_if_missing("events", "ticket_background_color", "VARCHAR(7) DEFAULT '#ffffff'")
|
|
_add_column_if_missing("events", "ticket_text_color", "VARCHAR(7) DEFAULT '#000000'")
|
|
_add_column_if_missing("events", "ticket_accent_color", "VARCHAR(7) DEFAULT '#0d6efd'")
|
|
_add_column_if_missing("events", "ticket_show_qr", "BOOLEAN DEFAULT 1")
|
|
_add_column_if_missing("events", "ticket_show_photo", "BOOLEAN DEFAULT 0")
|
|
_add_column_if_missing("events", "ticket_layout", "VARCHAR(20) DEFAULT 'portrait'")
|
|
_add_column_if_missing("events", "logo_filename", "VARCHAR(255)")
|
|
|
|
# Добавляем колонки ФИО/телефона в таблицу guests
|
|
_add_column_if_missing("guests", "last_name", "VARCHAR(255)")
|
|
_add_column_if_missing("guests", "first_name", "VARCHAR(255)")
|
|
_add_column_if_missing("guests", "middle_name", "VARCHAR(255)")
|
|
_add_column_if_missing("guests", "phone", "VARCHAR(50)")
|
|
|
|
# Версия сессии пользователя для инвалидации после смены пароля
|
|
_add_column_if_missing("users", "session_version", "INTEGER DEFAULT 0")
|
|
|
|
# Флаг отправки через локальный Outlook (MAPI)
|
|
_add_column_if_missing("smtp_settings", "use_local_outlook", "BOOLEAN DEFAULT 0")
|
|
# Флаг отправки через Exchange Web Services (EWS)
|
|
_add_column_if_missing("smtp_settings", "use_ews", "BOOLEAN DEFAULT 0")
|
|
_add_column_if_missing("smtp_settings", "ews_url", "VARCHAR(500)")
|
|
|
|
# Настройки IMAP для сохранения копий отправленных писем
|
|
_add_column_if_missing("smtp_settings", "imap_server", "VARCHAR(255)")
|
|
_add_column_if_missing("smtp_settings", "imap_port", "INTEGER DEFAULT 993")
|
|
_add_column_if_missing("smtp_settings", "imap_use_ssl", "BOOLEAN DEFAULT 1")
|
|
_add_column_if_missing("smtp_settings", "imap_username", "VARCHAR(255)")
|
|
_add_column_if_missing("smtp_settings", "imap_password", "VARCHAR(255)")
|
|
_add_column_if_missing("smtp_settings", "imap_folder", "VARCHAR(255)")
|
|
_add_column_if_missing("smtp_settings", "imap_save_copy", "BOOLEAN DEFAULT 0")
|
|
|
|
# Переносим файлы в папки по мероприятиям
|
|
_migrate_files_to_event_folders()
|
|
|
|
# Проверяем, есть ли колонка event_id в таблице guests
|
|
if _column_exists("guests", "event_id"):
|
|
db.session.commit()
|
|
return # Миграция гостей уже выполнена
|
|
|
|
# Создаём дефолтное мероприятие для существующих гостей
|
|
default_event = Event(
|
|
name=app.config.get("DEFAULT_EVENT_NAME", "Архивное мероприятие"),
|
|
description="Автоматически созданное мероприятие для существующих гостей",
|
|
is_active=True,
|
|
)
|
|
db.session.add(default_event)
|
|
db.session.flush()
|
|
|
|
# Добавляем колонку event_id
|
|
db.session.execute(text("ALTER TABLE guests ADD COLUMN event_id INTEGER REFERENCES events(id)"))
|
|
|
|
# Привязываем всех существующих гостей к дефолтному мероприятию
|
|
db.session.execute(
|
|
text("UPDATE guests SET event_id = :event_id"),
|
|
{"event_id": default_event.id},
|
|
)
|
|
|
|
# Добавляем колонку event_id в check_ins (опционально)
|
|
if not _column_exists("check_ins", "event_id"):
|
|
db.session.execute(text("ALTER TABLE check_ins ADD COLUMN event_id INTEGER REFERENCES events(id)"))
|
|
db.session.execute(
|
|
text("""
|
|
UPDATE check_ins
|
|
SET event_id = (
|
|
SELECT event_id FROM guests WHERE guests.id = check_ins.guest_id
|
|
)
|
|
""")
|
|
)
|
|
|
|
db.session.commit()
|
|
logger.info(f"Migration completed: existing guests moved to event '{default_event.name}' (id={default_event.id})")
|
|
except Exception as e:
|
|
db.session.rollback()
|
|
logger.warning(f"Migration skipped or failed: {e}")
|