366 lines
14 KiB
Python
366 lines
14 KiB
Python
# 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)
|
||
db.session.add(user)
|
||
db.session.commit()
|
||
|
||
# 텔레그램 알림 (인라인 버튼 포함)
|
||
message = (
|
||
f"🆕 <b>신규 가입 요청</b>\n\n"
|
||
f"<EFBFBD> 사용자: <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":
|
||
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")) |