diff --git a/app/api.py b/app/api.py index d50bb90..bfa2faa 100644 --- a/app/api.py +++ b/app/api.py @@ -8,6 +8,7 @@ from app.grad_2025.router import private_router as grad_2025_private_router from app.grad_2025.router import public_router as grad_2025_public_router from app.posts.router import router as posts_router +from app.slack.router import router as slack_router from app.tags.router import router as tags_router from app.users.router import router as users_router @@ -34,3 +35,4 @@ router.include_router(public_router) router.include_router(private_router) +router.include_router(slack_router, prefix="/slack", tags=["Slack"]) diff --git a/app/grad_2025/repository.py b/app/grad_2025/repository.py index 0fa2f62..0411696 100644 --- a/app/grad_2025/repository.py +++ b/app/grad_2025/repository.py @@ -16,3 +16,42 @@ async def list_univ_majors(self) -> list[Grad2025]: select(Grad2025).where(Grad2025.review_status == ReviewStatus.APPROVED) ) return list(result.all()) + + async def list_all_univ_majors(self) -> list[Grad2025]: + """전체 학교/전공 목록을 조회합니다 (승인 여부 무관).""" + result = await self.session.scalars( + select(Grad2025).order_by(Grad2025.review_status, Grad2025.name) + ) + return list(result.all()) + + async def find_by_name(self, name: str) -> Grad2025 | None: + """이름으로 학교/전공을 찾습니다.""" + result = await self.session.scalars( + select(Grad2025).where(Grad2025.name == name) + ) + return result.first() + + async def find_by_id(self, grad_id: int) -> Grad2025 | None: + """ID로 학교/전공을 찾습니다.""" + result = await self.session.scalars( + select(Grad2025).where(Grad2025.id == grad_id) + ) + return result.first() + + async def approve(self, grad_id: int) -> Grad2025 | None: + """학교/전공을 승인 처리합니다.""" + grad = await self.find_by_id(grad_id) + if grad is None: + return None + grad.review_status = ReviewStatus.APPROVED + await self.session.flush() + return grad + + async def update_name(self, grad_id: int, new_name: str) -> Grad2025 | None: + """학교/전공 이름을 변경합니다.""" + grad = await self.find_by_id(grad_id) + if grad is None: + return None + grad.name = new_name + await self.session.flush() + return grad diff --git a/app/slack/config.py b/app/slack/config.py index 8a92053..17dcda3 100644 --- a/app/slack/config.py +++ b/app/slack/config.py @@ -3,6 +3,10 @@ class SlackConfig(BaseConfig): SLACK_WEBHOOK_URL: str | None = None + SLACK_SIGNING_SECRET: str | None = None # 슬랙 요청 검증용 + SLACK_ALLOWED_USER_IDS: list[ + str + ] = [] # 허용된 사용자 ID 목록 (빈 리스트면 모두 허용) slack_settings = SlackConfig.create() diff --git a/app/slack/handlers.py b/app/slack/handlers.py new file mode 100644 index 0000000..f31c95d --- /dev/null +++ b/app/slack/handlers.py @@ -0,0 +1,132 @@ +from app.grad_2025.enums import ReviewStatus +from app.grad_2025.repository import Grad2025Repository +from app.utils.dependency import dependency + + +@dependency +class GradCommandHandler: + """슬랙 /grad 슬래시 커맨드 처리 핸들러""" + + grad_repository: Grad2025Repository + + async def handle_command(self, text: str) -> str: + """커맨드 텍스트를 파싱하고 적절한 핸들러를 호출합니다.""" + parts = text.strip().split(maxsplit=1) + if not parts: + return self._help_message() + + action = parts[0].lower() + args = parts[1] if len(parts) > 1 else "" + + match action: + case "list": + return await self.handle_list() + case "status": + return await self.handle_status(args) + case "approve": + return await self.handle_approve(args) + case "rename": + return await self.handle_rename(args) + case "help": + return self._help_message() + case _: + return f"❌ 알 수 없는 커맨드: `{action}`\n\n{self._help_message()}" + + async def handle_list(self) -> str: + """전체 학과 목록과 승인 여부를 반환합니다.""" + grads = await self.grad_repository.list_all_univ_majors() + + if not grads: + return "📚 등록된 학과가 없습니다." + + approved_count = sum( + 1 for g in grads if g.review_status == ReviewStatus.APPROVED + ) + total_count = len(grads) + + lines = [f"📚 *학과 목록* ({approved_count}/{total_count} 승인됨)\n"] + + for grad in grads: + if grad.review_status == ReviewStatus.APPROVED: + lines.append(f"✅ {grad.name}") + else: + lines.append(f"⏳ {grad.name} _(대기중)_") + + return "\n".join(lines) + + async def handle_status(self, name: str) -> str: + """특정 학과의 승인 여부를 반환합니다.""" + if not name: + return "❌ 학과명을 입력해주세요.\n사용법: `/grad status 홍익대학교 시각디자인`" + + grad = await self.grad_repository.find_by_name(name) + + if not grad: + return f"❌ `{name}` 학과를 찾을 수 없습니다." + + if grad.review_status == ReviewStatus.APPROVED: + return f"✅ *{grad.name}*\n상태: 승인됨" + else: + return f"⏳ *{grad.name}*\n상태: 대기중 (ID: {grad.id})" + + async def handle_approve(self, name: str) -> str: + """특정 학과를 승인 처리합니다.""" + if not name: + return "❌ 학과명을 입력해주세요.\n사용법: `/grad approve 홍익대학교 시각디자인`" + + grad = await self.grad_repository.find_by_name(name) + + if not grad: + return f"❌ `{name}` 학과를 찾을 수 없습니다." + + if grad.review_status == ReviewStatus.APPROVED: + return f"ℹ️ *{grad.name}*은(는) 이미 승인된 상태입니다." + + await self.grad_repository.approve(grad.id) + return f"✅ *{grad.name}*을(를) 승인했습니다!" + + async def handle_rename(self, args: str) -> str: + """학과 이름을 변경합니다.""" + # "기존이름 → 새이름" 또는 "기존이름 -> 새이름" 형식 파싱 + separators = ["→", "->", "=>"] + old_name = None + new_name = None + + for sep in separators: + if sep in args: + parts = args.split(sep, 1) + if len(parts) == 2: + old_name = parts[0].strip() + new_name = parts[1].strip() + break + + if not old_name or not new_name: + return ( + "❌ 올바른 형식으로 입력해주세요.\n" + "사용법: `/grad rename 기존이름 → 새이름`\n" + "예시: `/grad rename 홍익대학교 시각디자인 → 홍익대 시디`" + ) + + grad = await self.grad_repository.find_by_name(old_name) + + if not grad: + return f"❌ `{old_name}` 학과를 찾을 수 없습니다." + + # 새 이름이 이미 존재하는지 확인 + existing = await self.grad_repository.find_by_name(new_name) + if existing: + return f"❌ `{new_name}` 이름은 이미 사용 중입니다." + + await self.grad_repository.update_name(grad.id, new_name) + return f"✏️ 학과 이름을 변경했습니다.\n`{old_name}` → `{new_name}`" + + def _help_message(self) -> str: + """도움말 메시지를 반환합니다.""" + return ( + "📖 *사용 가능한 커맨드*\n\n" + "• `/grad list` - 전체 학과 목록과 승인 여부\n" + "• `/grad status 학과명` - 특정 학과의 승인 여부\n" + "• `/grad approve 학과명` - 학과 승인 처리\n" + "• `/grad rename 기존이름 → 새이름` - 학과 이름 변경\n" + "• `/grad help` - 이 도움말 보기" + ) diff --git a/app/slack/router.py b/app/slack/router.py new file mode 100644 index 0000000..ba0b483 --- /dev/null +++ b/app/slack/router.py @@ -0,0 +1,48 @@ +from typing import Annotated + +from fastapi import Form, Header, HTTPException, Request +from fastapi.responses import PlainTextResponse + +from app.core.router import create_router + +from .handlers import GradCommandHandler +from .service import is_user_allowed, verify_slack_signature + +router = create_router() + + +@router.post("/commands/grad", response_class=PlainTextResponse) +async def handle_grad_command( + request: Request, + handler: GradCommandHandler, + text: Annotated[str, Form()] = "", + user_id: Annotated[str, Form()] = "", + x_slack_request_timestamp: Annotated[str | None, Header()] = None, + x_slack_signature: Annotated[str | None, Header()] = None, +) -> str: + """ + /grad 슬래시 커맨드를 처리합니다. + + 사용 가능한 커맨드: + - /grad list - 전체 학과 목록과 승인 여부 + - /grad status 학과명 - 특정 학과의 승인 여부 + - /grad approve 학과명 - 학과 승인 처리 + - /grad rename 기존이름 → 새이름 - 학과 이름 변경 + - /grad help - 도움말 + """ + # 슬랙 서명 검증 + if x_slack_request_timestamp and x_slack_signature: + body = await request.body() + if not verify_slack_signature( + x_slack_request_timestamp, + body, + x_slack_signature, + ): + raise HTTPException(status_code=401, detail="Invalid Slack signature") + + # 사용자 권한 확인 + if not is_user_allowed(user_id): + return "❌ 이 커맨드를 실행할 권한이 없습니다." + + # 커맨드 처리 + return await handler.handle_command(text) diff --git a/app/slack/service.py b/app/slack/service.py index fd4da81..2eacffc 100644 --- a/app/slack/service.py +++ b/app/slack/service.py @@ -1,3 +1,7 @@ +import hashlib +import hmac +import time + import httpx from app.core.config import Environment, core_settings @@ -35,3 +39,41 @@ async def notify_new_univ_major(self, univ_major: str) -> None: phase = core_settings.ENVIRONMENT.value.upper() message = f"[{phase}] 🎓 새로운 학교/전공 검수 요청: *{univ_major}*" await self._send_message(message) + + +def verify_slack_signature( + timestamp: str, + body: bytes, + signature: str, +) -> bool: + """슬랙 요청의 서명을 검증합니다.""" + signing_secret = slack_settings.SLACK_SIGNING_SECRET + if not signing_secret: + # 로컬 개발 환경에서는 서명 검증 스킵 + return True + + # 요청이 5분 이상 된 경우 거부 (replay attack 방지) + if abs(time.time() - int(timestamp)) > 60 * 5: + return False + + # 서명 생성 + sig_basestring = f"v0:{timestamp}:{body.decode('utf-8')}" + my_signature = ( + "v0=" + + hmac.new( + signing_secret.encode(), + sig_basestring.encode(), + hashlib.sha256, + ).hexdigest() + ) + + return hmac.compare_digest(my_signature, signature) + + +def is_user_allowed(user_id: str) -> bool: + """사용자가 커맨드 실행 권한이 있는지 확인합니다.""" + allowed_users = slack_settings.SLACK_ALLOWED_USER_IDS + # 허용 목록이 비어있으면 모두 허용 + if not allowed_users: + return True + return user_id in allowed_users