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 "

Your OTP is {{OTP_CODE}}

" 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( """ Your OTP Code

{{APP_NAME}}

To complete your sign-in, please use the following one-time password (OTP).

{{OTP_CODE}}

This code is valid for 5 minutes.

Security Alert: For your protection, never share this code with anyone. Our team will never ask for your OTP over the phone or email.

If you did not request this code, you can safely ignore this email. Someone may have entered your email address by mistake.

© {{YEAR}} {{APP_NAME}}. All rights reserved.

MailOTP Guard — made with ❤️ by TechBitForge

""" ) # ================= 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)