Create app.py

app.py

@@ -0,0 +1,184 @@
+from flask import Flask, request, jsonify, render_template_string, redirect, session
+from flask_limiter import Limiter
+from flask_limiter.util import get_remote_address
+from datetime import datetime, timedelta
+import secrets
+import sqlite3
+import os
+
+app = Flask(__name__)
+app.secret_key = os.environ.get("FLASK_SECRET_KEY", "supersecret")  # Needed for session
+
+# --- Rate limiting ---
+limiter = Limiter(get_remote_address, app=app, default_limits=["10 per minute"])
+
+# --- Configurable credentials ---
+API_KEY = os.environ.get("OTP_API_KEY", "secretapikey123")
+ADMIN_USER = os.environ.get("ADMIN_USER", "admin")
+ADMIN_PASS = os.environ.get("ADMIN_PASS", "admin123")
+
+# --- DB init ---
+def init_db():
+    conn = sqlite3.connect("database.db")
+    c = conn.cursor()
+    c.execute('''CREATE TABLE IF NOT EXISTS otp (
+        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        code TEXT UNIQUE,
+        created_at TEXT,
+        expires_at TEXT,
+        used INTEGER DEFAULT 0,
+        used_at TEXT,
+        result TEXT,
+        ip TEXT
+    )''')
+    conn.commit()
+    conn.close()
+
+init_db()
+
+def generate_otp():
+    return str(secrets.randbelow(900000) + 100000)
+
+def insert_otp(code, created_at, expires_at, result="generated", ip=""):
+    conn = sqlite3.connect("database.db")
+    c = conn.cursor()
+    c.execute("INSERT INTO otp (code, created_at, expires_at, result, ip) VALUES (?, ?, ?, ?, ?)",
+              (code, created_at, expires_at, result, ip))
+    conn.commit()
+    conn.close()
+
+def mark_otp_used(code, result, ip):
+    now = datetime.utcnow().isoformat()
+    conn = sqlite3.connect("database.db")
+    c = conn.cursor()
+    c.execute("UPDATE otp SET used = 1, used_at = ?, result = ?, ip = ? WHERE code = ?",
+              (now, result, ip, code))
+    conn.commit()
+    conn.close()
+
+# --- Routes ---
+
+@app.route("/")
+def home():
+    return "✅ OTP API is running."
+
+@app.route("/generate-otp", methods=["POST"])
+@limiter.limit("3 per minute")
+def generate_api():
+    api_key = request.headers.get("X-API-Key")
+    if api_key != API_KEY:
+        return jsonify({"error": "Unauthorized"}), 401
+
+    otp = generate_otp()
+    now = datetime.utcnow()
+    expires = now + timedelta(minutes=5)
+    insert_otp(otp, now.isoformat(), expires.isoformat(), ip=request.remote_addr)
+
+    return jsonify({"otp": otp, "expires_at": expires.isoformat()})
+
+@app.route("/verify-otp", methods=["POST"])
+@limiter.limit("10 per minute")
+def verify():
+    data = request.json
+    otp = data.get("otp")
+    if not otp:
+        return jsonify({"success": False, "message": "No OTP provided"}), 400
+
+    now = datetime.utcnow()
+    conn = sqlite3.connect("database.db")
+    c = conn.cursor()
+    c.execute("SELECT used, expires_at FROM otp WHERE code = ?", (otp,))
+    row = c.fetchone()
+
+    if not row:
+        insert_otp(otp, now.isoformat(), now.isoformat(), result="invalid", ip=request.remote_addr)
+        return jsonify({"success": False, "message": "Invalid OTP"}), 401
+
+    used, expires_at = row
+    expires_at = datetime.fromisoformat(expires_at)
+
+    if used:
+        return jsonify({"success": False, "message": "OTP already used"}), 403
+    if now > expires_at:
+        mark_otp_used(otp, result="expired", ip=request.remote_addr)
+        return jsonify({"success": False, "message": "OTP expired"}), 403
+
+    mark_otp_used(otp, result="success", ip=request.remote_addr)
+    return jsonify({"success": True})
+
+@app.route("/export-log")
+def export_log():
+    api_key = request.headers.get("X-API-Key")
+    if api_key != API_KEY:
+        return jsonify({"error": "Unauthorized"}), 401
+
+    conn = sqlite3.connect("database.db")
+    c = conn.cursor()
+    c.execute("SELECT * FROM otp")
+    rows = c.fetchall()
+    conn.close()
+
+    return jsonify({
+        "columns": ["id", "code", "created_at", "expires_at", "used", "used_at", "result", "ip"],
+        "data": rows
+    })
+
+# --- UI with login ---
+
+@app.route("/login", methods=["GET", "POST"])
+def login():
+    error = ""
+    if request.method == "POST":
+        username = request.form.get("username")
+        password = request.form.get("password")
+        if username == ADMIN_USER and password == ADMIN_PASS:
+            session["logged_in"] = True
+            return redirect("/generate")
+        else:
+            error = "❌ Invalid login"
+    return render_template_string('''
+        <h2>🔐 Admin Login</h2>
+        <form method="post">
+            <input name="username" placeholder="Username" required><br><br>
+            <input name="password" type="password" placeholder="Password" required><br><br>
+            <input type="submit" value="Login">
+        </form>
+        <p>{{ error }}</p>
+    ''', error=error)
+
+@app.route("/generate", methods=["GET", "POST"])
+def generate_page():
+    if not session.get("logged_in"):
+        return redirect("/login")
+
+    message = ""
+    otp_data = None
+    if request.method == "POST":
+        otp = generate_otp()
+        now = datetime.utcnow()
+        expires = now + timedelta(minutes=5)
+        insert_otp(otp, now.isoformat(), expires.isoformat(), ip=request.remote_addr)
+        message = "✅ OTP generated!"
+        otp_data = {"otp": otp, "expires_at": expires.isoformat()}
+
+    return render_template_string("""
+        <h2>🎯 Generate OTP</h2>
+        <form method="post">
+            <input type="submit" value="Generate New OTP">
+        </form>
+        {% if message %}
+            <p><strong>{{ message }}</strong></p>
+        {% endif %}
+        {% if otp_data %}
+            <div>
+                <p><strong>OTP:</strong> {{ otp_data['otp'] }}</p>
+                <p><strong>Expires At:</strong> {{ otp_data['expires_at'] }}</p>
+            </div>
+        {% endif %}
+        <p><a href="/logout">Logout</a></p>
+    """, message=message, otp_data=otp_data)
+
+@app.route("/logout")
+def logout():
+    session.pop("logged_in", None)
+    return redirect("/login")