Spaces:

APINOW-service
/

OTP

Paused

File size: 12,806 Bytes

import os
import time
import random
import smtplib
from datetime import datetime
from functools import wraps
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart

# Flask & Extensions
from flask import Flask, request, jsonify
from flask_limiter import Limiter
from flask_cors import CORS
from flask_limiter.util import get_remote_address
from dotenv import load_dotenv

# REQUIRED FOR HUGGING FACE SPACES (Proxy Fix)
from werkzeug.middleware.proxy_fix import ProxyFix

# ================= LOAD ENV =================
load_dotenv()

app = Flask(__name__)

# ================= HUGGING FACE CONFIGURATION =================
# Hugging Face runs behind a proxy (Nginx). We must tell Flask to trust
# the X-Forwarded-For headers so Flask-Limiter sees the real user IP,
# not the proxy IP.
app.wsgi_app = ProxyFix(app.wsgi_app, x_for=1, x_proto=1, x_host=1, x_prefix=1)

CORS(app)

# ================= BASIC CONFIG =================
YOUR_WEB_APP_NAME = os.getenv("WEBAPPNAME", "xyzapp")
CURRENT_YEAR = str(datetime.now().year)
BASE_DIR = os.path.dirname(os.path.abspath(__file__))

# Ensure your Templates folder exists in the Space files
TEMPLATE_PATH = os.path.join(BASE_DIR, "Templates", "template1.html")
REMOVE_WATERMARK = False

# ================= RATE LIMIT =================
# With ProxyFix applied above, get_remote_address will now correctly
# grab the real user's IP address.
limiter = Limiter(
    key_func=get_remote_address,
    app=app,
    default_limits=["200 per hour"],
    storage_uri="memory://" # Explicitly use memory storage
)

# ================= SMTP CONFIG =================
SMTP_SERVER = "smtp.gmail.com"
SMTP_PORT = 587

GMAIL_EMAIL = os.getenv("GMAIL_EMAIL")
APP_PASSWORD = os.getenv("APP_PASSWORD")
FROM_EMAIL = os.getenv("FROM_EMAIL", GMAIL_EMAIL)

# ================= SECURITY CONFIG =================
OTP_EXPIRY_SECONDS = 300
MAX_ATTEMPTS = 5
LOCK_TIME = 600
RESEND_COOLDOWN = 60

IP_MAX_ATTEMPTS = 10
IP_BLOCK_TIME = 60

# ================= STORES (IN-MEMORY) =================
# Note: In-memory stores will reset if the Space restarts/sleeps.
otp_store = {}
ip_store = {}

# ================= HTML TEMPLATE =================
def load_template(path: str) -> str:
    # Fallback if file is missing to prevent crash
    if not os.path.exists(path):
        return "<html><body><h1>Your OTP is {{OTP_CODE}}</h1></body></html>"
        
    with open(path, "r", encoding="utf-8") as file:
        return file.read()

def apply_template_changes(template: str) -> str:
    template = template.replace("{{APP_NAME}}", YOUR_WEB_APP_NAME)
    template = template.replace("{{YEAR}}", CURRENT_YEAR)

    if REMOVE_WATERMARK:
        template = template.replace(
            "MailOTP Guard — made with ❤️ by TechBitForge",
            ""
        )
    return template

# Load template once on startup
HTML_TEMPLATE = apply_template_changes(
    """
    <!DOCTYPE html>
<html lang="en">
<head>
    <meta charset="UTF-8">
    <meta name="viewport" content="width=device-width, initial-scale=1.0">
    <meta http-equiv="X-UA-Compatible" content="IE=edge">
    <title>Your OTP Code</title>
    <style>
        /* Global Reset */
        body, table, td, a { -webkit-text-size-adjust: 100%; -ms-text-size-adjust: 100%; }
        table, td { mso-table-lspace: 0pt; mso-table-rspace: 0pt; }
        img { -ms-interpolation-mode: bicubic; border: 0; height: auto; line-height: 100%; outline: none; text-decoration: none; }
        table { border-collapse: collapse !important; }
        body { height: 100% !important; margin: 0 !important; padding: 0 !important; width: 100% !important; background-color: #f4f7f9; font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif; }

        /* Mobile Responsive */
        @media screen and (max-width: 600px) {
            .email-container { width: 100% !important; margin: auto !important; }
            .content-padding { padding: 20px !important; }
            .otp-code { font-size: 32px !important; letter-spacing: 8px !important; }
        }
    </style>
</head>
<body style="margin: 0; padding: 0; background-color: #f4f7f9;">
    <center>
        <table align="center" border="0" cellpadding="0" cellspacing="0" width="100%" style="max-width: 600px; margin: 40px auto; background-color: #ffffff; border-radius: 12px; overflow: hidden; box-shadow: 0 4px 12px rgba(0,0,0,0.05);">
            
            <tr>
                <td style="padding: 40px 40px 20px 40px; text-align: center;">
                    <h1 style="margin: 0; font-size: 24px; color: #1a1a1a; font-weight: 700; letter-spacing: -0.5px;">{{APP_NAME}}</h1>
                </td>
            </tr>

            <tr>
                <td class="content-padding" style="padding: 0 50px 40px 50px; text-align: center;">
                    <p style="margin: 0 0 24px 0; font-size: 16px; line-height: 24px; color: #4b5563;">
                        To complete your sign-in, please use the following one-time password (OTP).
                    </p>

                    <table align="center" border="0" cellpadding="0" cellspacing="0" style="margin: 0 auto;">
                        <tr>
                            <td style="background-color: #f8fafc; border: 1px solid #e2e8f0; border-radius: 8px; padding: 24px 40px;">
                                <span class="otp-code" style="font-family: 'Courier New', Courier, monospace; font-size: 42px; font-weight: 800; color: #2563eb; letter-spacing: 12px; display: block; margin-left: 12px;">{{OTP_CODE}}</span>
                            </td>
                        </tr>
                    </table>

                    <p style="margin: 24px 0 0 0; font-size: 14px; font-weight: 500; color: #64748b;">
                        This code is valid for <span style="color: #1e293b; font-weight: 600;">5 minutes</span>.
                    </p>
                </td>
            </tr>

            <tr>
                <td style="padding: 0 50px 40px 50px;">
                    <table border="0" cellpadding="0" cellspacing="0" width="100%" style="background-color: #fffbeb; border-radius: 8px; border-left: 4px solid #f59e0b;">
                        <tr>
                            <td style="padding: 16px; font-size: 13px; line-height: 20px; color: #92400e;">
                                <strong>Security Alert:</strong> For your protection, never share this code with anyone. Our team will never ask for your OTP over the phone or email.
                            </td>
                        </tr>
                    </table>
                </td>
            </tr>

            <tr>
                <td style="padding: 0 50px 40px 50px; text-align: center; border-top: 1px solid #f1f5f9;">
                    <p style="margin: 20px 0 8px 0; font-size: 12px; color: #94a3b8; line-height: 18px;">
                        If you did not request this code, you can safely ignore this email. Someone may have entered your email address by mistake.
                    </p>
                    <p style="margin: 0; font-size: 12px; color: #cbd5e1;">
                        &copy; {{YEAR}} {{APP_NAME}}. All rights reserved.
                    </p>
                    <p style="margin: 20px 0 0 0; font-size: 11px; color: #94a3b8; text-transform: uppercase; letter-spacing: 1px;">
                        MailOTP Guard &mdash; made with ❤️ by TechBitForge
                    </p>
                </td>
            </tr>
        </table>
    </center>
</body>
</html>
    """
)

# ================= IP BLOCK SYSTEM =================
def is_ip_blocked(ip):
    record = ip_store.get(ip)
    if not record:
        return False

    if time.time() > record["blocked_until"]:
        ip_store.pop(ip)
        return False

    return True

def register_ip_failure(ip):
    now = time.time()
    record = ip_store.setdefault(ip, {
        "attempts": 0,
        "blocked_until": 0
    })

    record["attempts"] += 1
    if record["attempts"] >= IP_MAX_ATTEMPTS:
        record["blocked_until"] = now + IP_BLOCK_TIME

# ================= SEND EMAIL =================
def send_email(email, otp):
    if not GMAIL_EMAIL or not APP_PASSWORD:
        print("Error: SMTP Credentials missing in Environment Variables")
        raise Exception("SMTP Config Missing")

    html = HTML_TEMPLATE.replace("{{OTP_CODE}}", otp)

    msg = MIMEMultipart()
    msg["From"] = FROM_EMAIL
    msg["To"] = email
    msg["Subject"] = f"{YOUR_WEB_APP_NAME} OTP Verification"
    msg.attach(MIMEText(html, "html"))

    with smtplib.SMTP(SMTP_SERVER, SMTP_PORT) as server:
        server.starttls()
        server.login(GMAIL_EMAIL, APP_PASSWORD)
        server.send_message(msg)

# ================= ROUTE: HOME =================
@app.route("/")
def home():
    return jsonify({"status": "running", "msg": f"Welcome to {YOUR_WEB_APP_NAME} API"})

# ================= ROUTE: SEND OTP =================
@app.route("/send-otp", methods=["POST"])
@limiter.limit("5 per minute")
def send_otp():
    # Behind ProxyFix, remote_addr is now safe
    ip = request.remote_addr

    if is_ip_blocked(ip):
        return jsonify({"error": "IP blocked for 1 minute"}), 429

    data = request.get_json(silent=True) or {}
    email = data.get("email")

    if not email:
        register_ip_failure(ip)
        return jsonify({"error": "Email required"}), 400

    now = time.time()
    otp = str(random.randint(100000, 999999))

    otp_store[email] = {
        "otp": otp,
        "expires": now + OTP_EXPIRY_SECONDS,
        "attempts": 0,
        "locked_until": 0,
        "last_sent": now
    }

    try:
        send_email(email, otp)
        return jsonify({"message": "OTP sent successfully"})
    except Exception as e:
        print(f"Mail Error: {e}") # Print error to HF Space logs
        register_ip_failure(ip)
        return jsonify({"error": "Failed to send email"}), 500

# ================= ROUTE: RESEND OTP =================
@app.route("/resend-otp", methods=["POST"])
@limiter.limit("3 per minute")
def resend_otp():
    ip = request.remote_addr

    if is_ip_blocked(ip):
        return jsonify({"error": "IP blocked for 1 minute"}), 429

    data = request.get_json(silent=True) or {}
    email = data.get("email")

    if not email:
        register_ip_failure(ip)
        return jsonify({"error": "Email required"}), 400

    record = otp_store.get(email)
    if not record:
        return jsonify({"error": "OTP not requested"}), 400

    now = time.time()

    if record["locked_until"] > now:
        return jsonify({"error": "Account locked"}), 429

    if now - record["last_sent"] < RESEND_COOLDOWN:
        return jsonify({
            "error": "Please wait before resending OTP",
            "retry_after": int(RESEND_COOLDOWN - (now - record["last_sent"]))
        }), 429

    otp = str(random.randint(100000, 999999))

    record.update({
        "otp": otp,
        "expires": now + OTP_EXPIRY_SECONDS,
        "attempts": 0,
        "last_sent": now
    })

    try:
        send_email(email, otp)
        return jsonify({"message": "OTP resent successfully"})
    except Exception:
        register_ip_failure(ip)
        return jsonify({"error": "Failed to resend OTP"}), 500

# ================= ROUTE: VERIFY OTP =================
@app.route("/verify-otp", methods=["POST"])
@limiter.limit("10 per minute")
def verify_otp():
    ip = request.remote_addr

    if is_ip_blocked(ip):
        return jsonify({"error": "IP blocked for 1 minute"}), 429

    data = request.get_json(silent=True) or {}
    email = data.get("email")
    user_otp = data.get("otp")

    if not email or not user_otp:
        register_ip_failure(ip)
        return jsonify({"error": "Missing fields"}), 400

    record = otp_store.get(email)
    if not record:
        register_ip_failure(ip)
        return jsonify({"error": "OTP not found"}), 400

    now = time.time()

    if record["locked_until"] > now:
        return jsonify({"error": "Account locked"}), 429

    if now > record["expires"]:
        otp_store.pop(email)
        return jsonify({"error": "OTP expired"}), 400

    if record["otp"] != user_otp:
        record["attempts"] += 1
        register_ip_failure(ip)

        if record["attempts"] >= MAX_ATTEMPTS:
            record["locked_until"] = now + LOCK_TIME
            return jsonify({"error": "Account locked for 10 minutes"}), 429

        return jsonify({
            "error": "Invalid OTP",
            "remaining_attempts": MAX_ATTEMPTS - record["attempts"]
        }), 400

    otp_store.pop(email)
    return jsonify({"message": "OTP verified successfully"})

# ================= RUN =================
if __name__ == "__main__":
    # Hugging Face Spaces default port is 7860
    app.run(host="0.0.0.0", port=7860, debug=False)