iDRAC_Info/backend/routes/auth.py

# backend/routes/auth.py
from __future__ import annotations

import logging
import threading
import asyncio
import secrets
from typing import Optional
from urllib.parse import urlparse, urljoin
from datetime import datetime

from flask import (
    Blueprint,
    render_template,
    redirect,
    url_for,
    flash,
    request,
    session,
    current_app,
)
from flask_login import login_user, logout_user, current_user, login_required

from backend.forms.auth_forms import RegistrationForm, LoginForm
from backend.models.user import User, db

# ── (선택) Telegram: 미설정이면 조용히 패스
try:
    from telegram import Bot, InlineKeyboardButton, InlineKeyboardMarkup
    from telegram.constants import ParseMode
except Exception:  # 라이브러리 미설치/미사용 환경
    Bot = None
    ParseMode = None
    InlineKeyboardButton = None
    InlineKeyboardMarkup = None

auth_bp = Blueprint("auth", __name__)


# ─────────────────────────────────────────────────────────────
# 유틸
# ─────────────────────────────────────────────────────────────
def _is_safe_url(target: str) -> bool:
    """로그인 후 next 파라미터의 안전성 확인(동일 호스트만 허용)."""
    ref = urlparse(request.host_url)
    test = urlparse(urljoin(request.host_url, target))
    return (test.scheme in ("http", "https")) and (ref.netloc == test.netloc)


def _notify(text: str, category: str = "system") -> None:
    """
    텔레그램 알림 전송
    - DB(TelegramBot)에 등록된 활성 봇들에게 전송
    - category: 'auth', 'activity', 'system' 등
    """
    try:
        from backend.models.telegram_bot import TelegramBot
        
        # 앱 컨텍스트 안에서 실행되므로 바로 DB 접근 가능
        try:
            bots = TelegramBot.query.filter_by(is_active=True).all()
        except Exception:
            db.create_all()
            bots = []

        # 1. DB에 봇이 없고, Config에 설정이 있는 경우 -> 자동 마이그레이션
        if not bots:
            cfg_token = (current_app.config.get("TELEGRAM_BOT_TOKEN") or "").strip()
            cfg_chat = (current_app.config.get("TELEGRAM_CHAT_ID") or "").strip()
            
            if cfg_token and cfg_chat:
                new_bot = TelegramBot(
                    name="기본 봇 (Config)",
                    token=cfg_token,
                    chat_id=cfg_chat,
                    is_active=True,
                    description="config.py 설정에서 자동 가져옴",
                    notification_types="auth,activity,system"
                )
                db.session.add(new_bot)
                db.session.commit()
                bots = [new_bot]
                current_app.logger.info("Telegram: Config settings migrated to DB.")

        if not bots:
            return

        # 카테고리 필터링
        target_bots = []
        for b in bots:
            allowed = (b.notification_types or "auth,activity,system").split(",")
            if category in allowed:
                target_bots.append(b)
        
        if not target_bots:
            return

        if not (Bot and ParseMode):
            current_app.logger.warning("Telegram: Library not installed.")
            return

        app = current_app._get_current_object()

        async def _send_to_bot(bot_obj, msg):
            try:
                t_bot = Bot(token=bot_obj.token)
                await t_bot.send_message(chat_id=bot_obj.chat_id, text=msg, parse_mode=ParseMode.HTML)
            except Exception as e:
                app.logger.error(f"Telegram fail ({bot_obj.name}): {e}")

        async def _send_all():
            tasks = [_send_to_bot(b, text) for b in target_bots]
            await asyncio.gather(*tasks)

        def _run_thread():
            try:
                asyncio.run(_send_all())
            except Exception as e:
                app.logger.error(f"Telegram async loop error: {e}")

        threading.Thread(target=_run_thread, daemon=True).start()

    except Exception as e:
        current_app.logger.error(f"Telegram notification error: {e}")


def _notify_with_buttons(text: str, buttons: list, category: str = "auth") -> None:
    """
    텔레그램 알림 전송 (인라인 버튼 포함)
    - buttons: [(text, callback_data), ...] 형식의 리스트
    """
    try:
        from backend.models.telegram_bot import TelegramBot
        
        try:
            bots = TelegramBot.query.filter_by(is_active=True).all()
        except Exception:
            db.create_all()
            bots = []

        if not bots:
            cfg_token = (current_app.config.get("TELEGRAM_BOT_TOKEN") or "").strip()
            cfg_chat = (current_app.config.get("TELEGRAM_CHAT_ID") or "").strip()
            
            if cfg_token and cfg_chat:
                new_bot = TelegramBot(
                    name="기본 봇 (Config)",
                    token=cfg_token,
                    chat_id=cfg_chat,
                    is_active=True,
                    description="config.py 설정에서 자동 가져옴",
                    notification_types="auth,activity,system"
                )
                db.session.add(new_bot)
                db.session.commit()
                bots = [new_bot]

        if not bots:
            return

        target_bots = []
        for b in bots:
            allowed = (b.notification_types or "auth,activity,system").split(",")
            if category in allowed:
                target_bots.append(b)
        
        if not target_bots:
            return

        if not (Bot and ParseMode and InlineKeyboardButton and InlineKeyboardMarkup):
            current_app.logger.warning("Telegram: Library not installed.")
            return

        # 인라인 키보드 생성
        keyboard = [[InlineKeyboardButton(btn[0], callback_data=btn[1]) for btn in buttons]]
        reply_markup = InlineKeyboardMarkup(keyboard)

        app = current_app._get_current_object()

        async def _send_to_bot(bot_obj, msg, markup):
            try:
                t_bot = Bot(token=bot_obj.token)
                await t_bot.send_message(
                    chat_id=bot_obj.chat_id,
                    text=msg,
                    parse_mode=ParseMode.HTML,
                    reply_markup=markup
                )
            except Exception as e:
                app.logger.error(f"Telegram fail ({bot_obj.name}): {e}")

        async def _send_all():
            tasks = [_send_to_bot(b, text, reply_markup) for b in target_bots]
            await asyncio.gather(*tasks)

        def _run_thread():
            try:
                asyncio.run(_send_all())
            except Exception as e:
                app.logger.error(f"Telegram async loop error: {e}")

        threading.Thread(target=_run_thread, daemon=True).start()

    except Exception as e:
        current_app.logger.error(f"Telegram notification with buttons error: {e}")


# ─────────────────────────────────────────────────────────────
# Blueprint 등록 훅
# ─────────────────────────────────────────────────────────────
def register_auth_routes(app):
    """app.py에서 register_routes(app, socketio) 호출 시 사용."""
    app.register_blueprint(auth_bp)

    @app.before_request
    def _global_hooks():
        # 1. 세션 갱신 (요청마다 세션 타임아웃 연장)
        session.modified = True

        # 2. 활동 알림 (로그인된 사용자)
        if current_user.is_authenticated:
            # 정적 리소스 및 불필요한 경로 제외
            if request.endpoint == 'static':
                return
            
            # 제외할 확장자/경로 (필요 시 추가)
            ignored_exts = ('.css', '.js', '.png', '.jpg', '.jpeg', '.gif', '.ico', '.woff', '.woff2')
            if request.path.lower().endswith(ignored_exts):
                return

            # 활동 내용 구성
            msg = (
                f"👣 <b>활동 감지</b>\n"
                f"👤 <code>{current_user.username}</code>\n"
                f"📍 <code>{request.method} {request.path}</code>"
            )
            _notify(msg, category="activity")


# ─────────────────────────────────────────────────────────────
# 회원가입
# ─────────────────────────────────────────────────────────────
@auth_bp.route("/register", methods=["GET", "POST"])
def register():
    if current_user.is_authenticated:
        current_app.logger.info("REGISTER: already auth → /index")
        return redirect(url_for("main.index"))

    form = RegistrationForm()
    if form.validate_on_submit():
        # 모델 내부에서 email/username 정규화됨(find_by_*)
        if User.find_by_email(form.email.data):
            flash("이미 등록된 이메일입니다.", "warning")
            current_app.logger.info("REGISTER: dup email %s", form.email.data)
            return render_template("register.html", form=form)

        if User.find_by_username(form.username.data):
            flash("이미 사용 중인 사용자명입니다.", "warning")
            current_app.logger.info("REGISTER: dup username %s", form.username.data)
            return render_template("register.html", form=form)

        # 승인 토큰 생성
        approval_token = secrets.token_urlsafe(32)
        
        user = User(
            username=form.username.data,
            email=form.email.data,
            is_active=False,
            is_approved=False,
            approval_token=approval_token
        )
        user.set_password(form.password.data)
        
        try:
            db.session.add(user)
            db.session.commit()
        except Exception as e:
            db.session.rollback()
            current_app.logger.error("REGISTER: DB commit failed: %s", e)
            flash("회원가입 처리 중 오류가 발생했습니다. (DB Error)", "danger")
            return render_template("register.html", form=form)

        # 텔레그램 알림 (인라인 버튼 포함)
        message = (
            f"🆕 <b>신규 가입 요청</b>\n\n"
            f"� 사용자: <code>{user.username}</code>\n"
            f"📧 이메일: <code>{user.email}</code>\n"
            f"🕐 신청시간: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}"
        )
        
        buttons = [
            ("✅ 승인", f"approve_{approval_token}"),
            ("❌ 거부", f"reject_{approval_token}")
        ]
        
        _notify_with_buttons(message, buttons, category="auth")
        
        current_app.logger.info("REGISTER: created id=%s email=%s token=%s", user.id, user.email, approval_token[:10])
        flash("회원가입이 완료되었습니다. 관리자의 승인을 기다려주세요.", "success")
        return redirect(url_for("auth.login"))
    else:
        if request.method == "POST":
            # 폼 검증 실패 에러를 Flash 메시지로 출력
            for field_name, errors in form.errors.items():
                for error in errors:
                    # 필드 객체 가져오기 (라벨 텍스트 확인용)
                    field = getattr(form, field_name, None)
                    label = field.label.text if field else field_name
                    flash(f"{label}: {error}", "warning")
            current_app.logger.info("REGISTER: form errors=%s", form.errors)

    return render_template("register.html", form=form)


# ─────────────────────────────────────────────────────────────
# 로그인
# ─────────────────────────────────────────────────────────────
@auth_bp.route("/login", methods=["GET", "POST"])
def login():
    if current_user.is_authenticated:
        current_app.logger.info("LOGIN: already auth → /index")
        return redirect(url_for("main.index"))

    form = LoginForm()
    if form.validate_on_submit():
        current_app.logger.info("LOGIN: form ok email=%s", form.email.data)
        user: Optional[User] = User.find_by_email(form.email.data)

        if not user:
            flash("이메일 또는 비밀번호가 올바르지 않습니다.", "danger")
            current_app.logger.info("LOGIN: user not found")
            return render_template("login.html", form=form)

        pass_ok = user.check_password(form.password.data)
        current_app.logger.info(
            "LOGIN: found id=%s active=%s approved=%s pass_ok=%s",
            user.id, user.is_active, user.is_approved, pass_ok
        )

        if not pass_ok:
            flash("이메일 또는 비밀번호가 올바르지 않습니다.", "danger")
            return render_template("login.html", form=form)

        if not user.is_approved:
            flash("계정이 아직 승인되지 않았습니다. 관리자의 승인을 기다려주세요.", "warning")
            return render_template("login.html", form=form)

        if not user.is_active:
            flash("계정이 비활성화되었습니다. 관리자에게 문의하세요.", "warning")
            return render_template("login.html", form=form)

        # 성공
        login_user(user, remember=form.remember.data)
        session.permanent = True
        _notify(f"🔐 <b>로그인 성공</b>\n👤 <code>{user.username}</code>", category="auth")
        current_app.logger.info("LOGIN: SUCCESS → redirect")

        nxt = request.args.get("next")
        if nxt and _is_safe_url(nxt):
            return redirect(nxt)
        return redirect(url_for("main.index"))
    else:
        if request.method == "POST":
            current_app.logger.info("LOGIN: form errors=%s", form.errors)

    return render_template("login.html", form=form)


# ─────────────────────────────────────────────────────────────
# 로그아웃
# ─────────────────────────────────────────────────────────────
@auth_bp.route("/logout", methods=["GET"])
@login_required
def logout():
    if current_user.is_authenticated:
        current_app.logger.info("LOGOUT: user=%s", current_user.username)
        _notify(f"🚪 <b>로그아웃</b>\n👤 <code>{current_user.username}</code>", category="auth")
    logout_user()
    flash("정상적으로 로그아웃 되었습니다.", "success")
    return redirect(url_for("auth.login"))