site/WESP_REL/app/routes/auth.py

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError
from werkzeug.security import check_password_hash, generate_password_hash
from flask import Blueprint, current_app, jsonify, request, session

from app import db
from app.models import WebUser
from config import Config
from app.routes.auth_decorators import can_access_lab, require_auth

bp = Blueprint("auth", __name__, url_prefix="/api/auth")

_PASSWORD_MIN_LEN = 8
_USER_RULES_PAYLOAD = {
    "password_min_len": _PASSWORD_MIN_LEN,
}


def load_credentials():
    """Получение учетных данных из переменных окружения."""
    return (
        current_app.config.get("AUTH_LOGIN", getattr(Config, "AUTH_LOGIN", "admin")),
        current_app.config.get(
            "AUTH_PASSWORD", getattr(Config, "AUTH_PASSWORD", "admin")
        ),
    )


def _ensure_default_superuser() -> None:
    saved_login, saved_password = load_credentials()
    login_key = saved_login.strip()
    if db.session.scalar(select(WebUser.id).where(WebUser.login == login_key).limit(1)):
        return
    if db.session.scalar(select(WebUser.id).limit(1)):
        return

    user = WebUser(
        login=login_key,
        password_hash=generate_password_hash(saved_password),
        is_superuser=True,
    )
    db.session.add(user)
    try:
        db.session.commit()
    except IntegrityError:
        db.session.rollback()


def _get_user_by_login(login: str) -> WebUser | None:
    return db.session.execute(
        select(WebUser).where(WebUser.login == login.strip())
    ).scalar_one_or_none()


def save_credentials(login: str, password: str) -> None:
    """Смена учетных данных через API отключена (env-only)."""
    _ = (login, password)
    raise RuntimeError(
        "Смена учетных данных через API отключена. Используйте WESP_AUTH_LOGIN/WESP_AUTH_PASSWORD."
    )


@bp.get("/ping")
def auth_ping():
    return jsonify({"status": "ok"}), 200


@bp.post("/login")
def authenticate_login():
    data = request.json or {}
    login = (data.get("login") or "").strip()
    password = data.get("password") or ""

    if not login:
        return jsonify({"status": "error", "message": "Не указан логин"}), 400
    if not password:
        return jsonify({"status": "error", "message": "Не указан пароль"}), 400

    _ensure_default_superuser()
    user = _get_user_by_login(login)
    if user is not None and check_password_hash(user.password_hash, password):
        remember = bool(data.get("remember"))
        session["authenticated"] = True
        session["user_login"] = user.login
        session["is_superuser"] = bool(user.is_superuser)
        session["lab_access"] = bool(user.is_superuser or user.lab_access)
        session["remember_login"] = remember
        if remember:
            session.permanent = True
        else:
            session.permanent = False
        return jsonify(
            {
                "status": "success",
                "message": "Авторизация успешна",
                "authenticated": True,
            }
        )

    return (
        jsonify(
            {
                "status": "error",
                "message": "Неверный логин или пароль",
                "authenticated": False,
            }
        ),
        401,
    )


@bp.post("/logout")
def logout():
    session.clear()
    return jsonify({"status": "success", "message": "Выход выполнен успешно"})


@bp.get("/check")
def check_auth():
    authenticated = session.get("authenticated", False)
    user_login = session.get("user_login", "")
    remember_login = bool(session.get("remember_login"))
    is_superuser = bool(session.get("is_superuser", False))
    can_lab = can_access_lab() if authenticated else False
    if authenticated:
        session["lab_access"] = can_lab
    return jsonify(
        {
            "status": "success",
            "authenticated": authenticated,
            "user_login": user_login,
            "remember_login": remember_login,
            "is_superuser": is_superuser,
            "can_lab": can_lab,
            "user_rules": _USER_RULES_PAYLOAD,
        }
    )


@bp.get("/get_current_credentials")
def get_current_credentials():
    current_login = (session.get("user_login") or "").strip()
    if not current_login:
        saved_login, _saved_password = load_credentials()
        current_login = saved_login
    return jsonify(
        {
            "status": "success",
            "login": current_login,
            # Никогда не возвращаем пароль в API.
            "password": "",
            "user_rules": _USER_RULES_PAYLOAD,
        }
    )


@bp.post("/change_credentials")
@require_auth
def change_credentials():
    from app.routes.admin import (
        _WEB_USER_PASSWORD_MIN_LEN,
        _reject_if_weak_login_or_password,
        _valid_login,
    )

    data = request.json or {}
    old_login = (data.get("old_login") or "").strip()
    old_password = data.get("old_password") or ""
    new_login = (data.get("new_login") or "").strip()
    new_password = data.get("new_password") or ""
    confirm_password = data.get("confirm_password") or new_password

    session_login = (session.get("user_login") or "").strip()
    if not session_login:
        return jsonify({"status": "error", "message": "Сессия не найдена"}), 401

    user = _get_user_by_login(session_login)
    if user is None:
        return jsonify({"status": "error", "message": "Пользователь не найден"}), 404

    if old_login != user.login:
        return (
            jsonify({"status": "error", "message": "Неверный текущий логин или пароль"}),
            401,
        )
    if not check_password_hash(user.password_hash, old_password):
        return (
            jsonify({"status": "error", "message": "Неверный текущий логин или пароль"}),
            401,
        )

    if not new_login:
        return jsonify({"status": "error", "message": "Новый логин не может быть пустым"}), 400
    if not new_password:
        return jsonify({"status": "error", "message": "Новый пароль не может быть пустым"}), 400
    if new_password != confirm_password:
        return jsonify({"status": "error", "message": "Новые пароли не совпадают"}), 400

    if not _valid_login(new_login):
        return (
            jsonify(
                {
                    "status": "error",
                    "message": "Логин не длиннее 64 символов",
                }
            ),
            400,
        )

    if len(new_password) < _WEB_USER_PASSWORD_MIN_LEN:
        return (
            jsonify(
                {
                    "status": "error",
                    "message": f"Пароль должен быть не короче {_WEB_USER_PASSWORD_MIN_LEN} символов",
                }
            ),
            400,
        )

    err_body, err_code = _reject_if_weak_login_or_password(new_login, new_password)
    if err_body is not None:
        return jsonify(err_body), err_code

    if new_login != user.login:
        existing = _get_user_by_login(new_login)
        if existing is not None and existing.id != user.id:
            return (
                jsonify({"status": "error", "message": "Пользователь с таким логином уже существует"}),
                409,
            )

    user.login = new_login
    user.password_hash = generate_password_hash(new_password)
    try:
        db.session.commit()
    except IntegrityError:
        db.session.rollback()
        return (
            jsonify({"status": "error", "message": "Пользователь с таким логином уже существует"}),
            409,
        )

    session["user_login"] = new_login
    return jsonify({"status": "success", "message": "Учетные данные обновлены"})