Spaces:

APINOW-service
/

OTP

Paused

App Files Files Community

APINOW-service commited on Jan 27

Commit

feeca53

verified ·

1 Parent(s): 3fc2398

Create app.py

Browse files

Files changed (1) hide show

app.py +277 -0

app.py ADDED Viewed

	@@ -0,0 +1,277 @@

+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(load_template(TEMPLATE_PATH))
+# ================= 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)