# secure_otp_service.py
from flask import Flask, jsonify, request, render_template_string
from datetime import datetime, timedelta
import random
import string
import logging
from functools import wraps
app = Flask(__name__)
app.logger.setLevel(logging.INFO)
# Configuration
CONFIG = {
"OTP_LENGTH": 6,
"VN_LENGTH": 10,
"EXPIRE_SECONDS": 300,
"MAX_ATTEMPTS": 3,
"RATE_LIMIT": 5 # Max requests per minute
}
# In-memory storage with expiration tracking
storage = {
"virtual_numbers": {},
"rate_limits": {}
}
HTML_TEMPLATE = """
Secure OTP Service
✨ Generate Virtual Number
"""
# Utility functions
def generate_secure_code(length, chars):
return ''.join(random.SystemRandom().choice(chars) for _ in range(length))
def cleanup_expired_entries():
now = datetime.now()
expired = [vn for vn, data in storage['virtual_numbers'].items() if data['expires_at'] < now]
for vn in expired:
del storage['virtual_numbers'][vn]
def check_rate_limit(ip):
now = datetime.now()
window_start = now - timedelta(minutes=1)
requests = [t for t in storage['rate_limits'].get(ip, []) if t > window_start]
if len(requests) >= CONFIG['RATE_LIMIT']:
return False
storage['rate_limits'][ip] = requests + [now]
return True
# Decorators
def rate_limit(f):
@wraps(f)
def wrapper(*args, **kwargs):
ip = request.remote_addr
if not check_rate_limit(ip):
return jsonify({"error": "Rate limit exceeded"}), 429
return f(*args, **kwargs)
return wrapper
# API endpoints
@app.route('/')
def home():
return render_template_string(HTML_TEMPLATE)
@app.route('/api/generate', methods=['POST'])
@rate_limit
def generate_virtual_number():
try:
cleanup_expired_entries()
ip = request.remote_addr
while True:
vn = generate_secure_code(CONFIG['VN_LENGTH'], string.digits)
if vn not in storage['virtual_numbers']:
break
otp = generate_secure_code(CONFIG['OTP_LENGTH'], string.digits)
expires_at = datetime.now() + timedelta(seconds=CONFIG['EXPIRE_SECONDS'])
storage['virtual_numbers'][vn] = {
'otp': otp,
'expires_at': expires_at,
'created_at': datetime.now(),
'attempts': 0,
'ip': ip
}
app.logger.info(f"Generated virtual number: {vn}")
return jsonify({
"virtual_number": vn,
"otp": otp,
"expires_in": CONFIG['EXPIRE_SECONDS']
})
except Exception as e:
app.logger.error(f"Generation error: {str(e)}")
return jsonify({"error": "Internal server error"}), 500
@app.route('/api/validate', methods=['POST'])
@rate_limit
def validate_otp():
try:
data = request.get_json()
if not data or 'virtual_number' not in data or 'otp' not in data:
return jsonify({"error": "Invalid request format"}), 400
vn = data['virtual_number']
otp_attempt = data['otp']
cleanup_expired_entries()
record = storage['virtual_numbers'].get(vn)
if not record:
return jsonify({"valid": False, "message": "Invalid or expired virtual number"})
if record['attempts'] >= CONFIG['MAX_ATTEMPTS']:
del storage['virtual_numbers'][vn]
return jsonify({"valid": False, "message": "Maximum attempts exceeded"})
storage['virtual_numbers'][vn]['attempts'] += 1
if record['otp'] == otp_attempt:
del storage['virtual_numbers'][vn]
return jsonify({"valid": True, "message": "OTP validated successfully"})
return jsonify({
"valid": False,
"message": f"Invalid OTP ({CONFIG['MAX_ATTEMPTS'] - record['attempts']} attempts remaining)"
})
except Exception as e:
app.logger.error(f"Validation error: {str(e)}")
return jsonify({"error": "Internal server error"}), 500
if __name__ == '__main__':
app.run(host='0.0.0.0', port=5000, debug=False)
pip install flask
python secure_otp_service.py