|
| 1 | +import redis |
| 2 | +import random |
| 3 | +import string |
| 4 | +import time |
| 5 | +from typing import List, Tuple |
| 6 | +from datetime import datetime, timedelta |
| 7 | +from email.mime.text import MIMEText |
| 8 | +import smtplib |
| 9 | +import json |
| 10 | +import os |
| 11 | + |
| 12 | +# ------------------------------- |
| 13 | +# Redis 接続 (初期化は後で) |
| 14 | +# ------------------------------- |
| 15 | +r = None |
| 16 | +prefix = "" # app から渡す |
| 17 | + |
| 18 | +def init_redis(redis_conn, key_prefix: str = ""): |
| 19 | + global r, prefix |
| 20 | + r = redis_conn |
| 21 | + prefix = key_prefix |
| 22 | + |
| 23 | +# ------------------------------- |
| 24 | +# メール許可・所属 |
| 25 | +# ------------------------------- |
| 26 | +with open("config/allowed_emails.json", encoding="utf-8") as f: |
| 27 | + _ALLOWED = json.load(f) |
| 28 | + |
| 29 | +def is_allowed(email: str) -> bool: |
| 30 | + return email in _ALLOWED |
| 31 | + |
| 32 | +def get_affiliations(email: str) -> List[str]: |
| 33 | + if not is_allowed(email): |
| 34 | + return [] |
| 35 | + aff = _ALLOWED[email] |
| 36 | + if isinstance(aff, list): |
| 37 | + return [str(a).strip() for a in aff if str(a).strip()] |
| 38 | + if isinstance(aff, str): |
| 39 | + return [a.strip() for a in aff.split(",") if a.strip()] |
| 40 | + return [] |
| 41 | + |
| 42 | +# ------------------------------- |
| 43 | +# Gmail SMTP 設定 |
| 44 | +# ------------------------------- |
| 45 | +SMTP_SERVER = os.environ.get("SMTP_SERVER", "smtp.gmail.com") |
| 46 | +SMTP_PORT = int(os.environ.get("SMTP_PORT", 587)) |
| 47 | +SMTP_USER = os.environ.get("SMTP_USER") |
| 48 | +SMTP_PASS = os.environ.get("SMTP_PASS") |
| 49 | + |
| 50 | +# ------------------------------- |
| 51 | +# OTP 設定 |
| 52 | +# ------------------------------- |
| 53 | +OTP_TTL_SECONDS = 5 * 60 |
| 54 | +MAX_OTP_ATTEMPTS = 5 |
| 55 | + |
| 56 | +def generate_otp_code(length=6) -> str: |
| 57 | + return ''.join(random.choices(string.digits, k=length)) |
| 58 | + |
| 59 | +def send_otp(email: str) -> Tuple[bool, str]: |
| 60 | + """OTP を生成して Redis に保存してメール送信""" |
| 61 | + public_message = "メールを確認してください。" |
| 62 | + if not is_allowed(email): |
| 63 | + return True, public_message |
| 64 | + |
| 65 | + code = generate_otp_code() |
| 66 | + now = int(time.time()) |
| 67 | + key = f"{prefix}:otp:{email}" |
| 68 | + |
| 69 | + r.hmset(key, { |
| 70 | + "code": code, |
| 71 | + "expires_at": now + OTP_TTL_SECONDS, |
| 72 | + "attempts_left": MAX_OTP_ATTEMPTS, |
| 73 | + }) |
| 74 | + r.expire(key, OTP_TTL_SECONDS) |
| 75 | + |
| 76 | + # メール送信 |
| 77 | + msg = MIMEText( |
| 78 | + f"あなたの認証コード (OTP) は {code} です。\n" |
| 79 | + f"有効期限は {datetime.fromtimestamp(now + OTP_TTL_SECONDS).strftime('%H:%M:%S')} までです。" |
| 80 | + ) |
| 81 | + msg["Subject"] = "OTP 認証コード" |
| 82 | + msg["From"] = SMTP_USER |
| 83 | + msg["To"] = email |
| 84 | + |
| 85 | + try: |
| 86 | + with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server: |
| 87 | + server.starttls() |
| 88 | + server.login(SMTP_USER, SMTP_PASS) |
| 89 | + server.sendmail(SMTP_USER, [email], msg.as_string()) |
| 90 | + except Exception as e: |
| 91 | + return False, f"メール送信失敗: {e}" |
| 92 | + |
| 93 | + return True, public_message |
| 94 | + |
| 95 | +def verify_otp(email: str, code: str) -> bool: |
| 96 | + key = f"{prefix}:otp:{email}" |
| 97 | + data = r.hgetall(key) |
| 98 | + now = int(time.time()) |
| 99 | + if not data: |
| 100 | + return False |
| 101 | + |
| 102 | + if now > int(data.get("expires_at", 0)): |
| 103 | + r.delete(key) |
| 104 | + return False |
| 105 | + |
| 106 | + attempts_left = int(data.get("attempts_left", MAX_OTP_ATTEMPTS)) |
| 107 | + if attempts_left <= 0: |
| 108 | + r.delete(key) |
| 109 | + return False |
| 110 | + |
| 111 | + if code == data.get("code"): |
| 112 | + r.delete(key) |
| 113 | + return True |
| 114 | + else: |
| 115 | + attempts_left -= 1 |
| 116 | + if attempts_left <= 0: |
| 117 | + r.delete(key) |
| 118 | + else: |
| 119 | + r.hset(key, "attempts_left", attempts_left) |
| 120 | + return False |
| 121 | + |
| 122 | +def invalidate_otp(email: str): |
| 123 | + r.delete(f"{prefix}:otp:{email}") |
| 124 | + |
0 commit comments