I'll create a Python POST request handler for a login form. This will be a complete, production-ready example using Flask with security best practices.
from flask import Flask, request, jsonify, make_response
from functools import wraps
import re
import hashlib
import secrets
import time
from datetime import datetime, timedelta
app = Flask(__name__)
# Configuration
app.config['SECRET_KEY'] = secrets.token_hex(32) # In production, set this as env variable
app.config['MAX_LOGIN_ATTEMPTS'] = 5
app.config['LOCKOUT_DURATION'] = 300 # 5 minutes in seconds
# Simulated user database (in production, use proper database)
# Passwords should be stored as properly salted hashes
USERS_DB = {
'admin': {
'password_hash': hashlib.sha256('admin123'.encode()).hexdigest(), # Demo only - use bcrypt in production
'salt': 'random_salt_here',
'role': 'admin',
'active': True
},
'user': {
'password_hash': hashlib.sha256('user123'.encode()).hexdigest(),
'salt': 'another_salt',
'role': 'user',
'active': True
}
}
# Rate limiting storage (in production, use Redis)
login_attempts = {}
def get_client_ip():
"""Get client IP address, considering proxies"""
if request.headers.get('X-Forwarded-For'):
return request.headers.get('X-Forwarded-For').split(',')[0].strip()
return request.remote_addr
def check_rate_limit(ip_address):
"""Check if IP is rate limited due to failed attempts"""
now = time.time()
if ip_address in login_attempts:
attempts = login_attempts[ip_address]
# Clean old attempts outside window
attempts['times'] = [t for t in attempts['times'] if now - t < 3600]
# Check lockout
if attempts.get('locked_until') and now < attempts['locked_until']:
remaining = int(attempts['locked_until'] - now)
return False, f"Account locked. Try again in {remaining} seconds."
# Check attempt count
if len(attempts['times']) >= app.config['MAX_LOGIN_ATTEMPTS']:
attempts['locked_until'] = now + app.config['LOCKOUT_DURATION']
return False, f"Too many attempts. Account locked for 5 minutes."
return True, None
def record_attempt(ip_address, success=False):
"""Record login attempt"""
now = time.time()
if ip_address not in login_attempts:
login_attempts[ip_address] = {'times': [], 'locked_until': None}
if not success:
login_attempts[ip_address]['times'].append(now)
else:
# Clear attempts on success
login_attempts[ip_address] = {'times': [], 'locked_until': None}
def validate_email(email):
"""Validate email format"""
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(pattern, email) is not None
def sanitize_input(data):
"""Basic input sanitization"""
if not isinstance(data, str):
return None
# Remove null bytes and control characters
sanitized = ''.join(char for char in data if ord(char) > 31 and ord(char) != 127)
return sanitized.strip()
def verify_password(stored_hash, provided_password, salt):
"""Verify password against stored hash (demo - use bcrypt/argon2 in production)"""
# In production: use bcrypt.checkpw(provided_password.encode(), stored_hash)
hash_obj = hashlib.sha256((provided_password + salt).encode())
return hash_obj.hexdigest() == stored_hash
def generate_session_token():
"""Generate secure session token"""
return secrets.token_urlsafe(32)
@app.route('/login', methods=['POST'])
def login():
"""
Handle login POST request
Expected JSON payload:
{
"username": "string",
"password": "string",
"remember_me": false // optional
}
Returns:
- 200: Success with session token
- 400: Bad request (validation error)
- 401: Unauthorized (invalid credentials)
- 429: Too many requests (rate limited)
- 500: Server error
"""
try:
# Get client IP for rate limiting
client_ip = get_client_ip()
# Check rate limiting
allowed, message = check_rate_limit(client_ip)
if not allowed:
response = make_response(jsonify({
'success': False,
'error': 'rate_limited',
'message': message
}), 429)
response.headers['Retry-After'] = str(app.config['LOCKOUT_DURATION'])
return response
# Parse JSON payload
if not request.is_json:
return jsonify({
'success': False,
'error': 'invalid_content_type',
'message': 'Content-Type must be application/json'
}), 400
data = request.get_json()
# Validate required fields
if not data:
return jsonify({
'success': False,
'error': 'missing_body',
'message': 'Request body is required'
}), 400
# Extract and sanitize inputs
username = sanitize_input(data.get('username', ''))
password = data.get('password', '') # Don't sanitize password, just validate presence
remember_me = data.get('remember_me', False)
# Validation checks
errors = []
if not username:
errors.append('Username is required')
elif len(username) < 3 or len(username) > 32:
errors.append('Username must be between 3 and 32 characters')
if not password:
errors.append('Password is required')
elif len(password) < 6:
errors.append('Password must be at least 6 characters')
if errors:
return jsonify({
'success': False,
'error': 'validation_error',
'message': 'Validation failed',
'details': errors
}), 400
# Check if user exists and is active
user = USERS_DB.get(username)
if not user:
# Generic error message to prevent user enumeration
record_attempt(client_ip, success=False)
return jsonify({
'success': False,
'error': 'invalid_credentials',
'message': 'Invalid username or password'
}), 401
if not user.get('active', True):
return jsonify({
'success': False,
'error': 'account_disabled',
'message': 'Account has been disabled'
}), 403
# Verify password
if not verify_password(user['password_hash'], password, user['salt']):
record_attempt(client_ip, success=False)
return jsonify({
'success': False,
'error': 'invalid_credentials',
'message': 'Invalid username or password'
}), 401
# Successful authentication
record_attempt(client_ip, success=True)
# Generate session
session_token = generate_session_token()
# Calculate expiration
if remember_me:
expires = datetime.utcnow() + timedelta(days=30)
max_age = 30 * 24 * 60 * 60 # 30 days
else:
expires = datetime.utcnow() + timedelta(hours=24)
max_age = 24 * 60 * 60 # 24 hours
# Build response
response_data = {
'success': True,
'message': 'Login successful',
'user': {
'username': username,
'role': user['role']
},
'session': {
'token': session_token,
'expires_at': expires.isoformat()
}
}
response = make_response(jsonify(response_data), 200)
# Set secure cookie (in production, use HttpOnly, Secure, SameSite)
response.set_cookie(
'session_token',
session_token,
max_age=max_age,
expires=expires,
httponly=True, # Prevent XSS access
secure=True, # HTTPS only in production
samesite='Lax' # CSRF protection
)
# Security headers
response.headers['X-Content-Type-Options'] = 'nosniff'
response.headers['X-Frame-Options'] = 'DENY'
response.headers['X-XSS-Protection'] = '1; mode=block'
return response
except Exception as e:
# Log error (in production, use proper logging)
app.logger.error(f"Login error: {str(e)}")
return jsonify({
'success': False,
'error': 'server_error',
'message': 'An unexpected error occurred'
}), 500
@app.route('/logout', methods=['POST'])
def logout():
"""Handle logout - clear session"""
response = make_response(jsonify({
'success': True,
'message': 'Logged out successfully'
}), 200)
# Clear cookie
response.set_cookie('session_token', '', expires=0)
return response
@app.route('/register', methods=['POST'])
def register():
"""
Handle registration (simplified example)
"""
try:
data = request.get_json()
username = sanitize_input(data.get('username', ''))
password = data.get('password', '')
email = sanitize_input(data.get('email', ''))
# Validation
if username in USERS_DB:
return jsonify({
'success': False,
'error': 'username_exists',
'message': 'Username already taken'
}), 409
if not validate_email(email):
return jsonify({
'success': False,
'error': 'invalid_email',
'message': 'Invalid email format'
}), 400
# In production: hash with bcrypt/argon2
salt = secrets.token_hex(16)
password_hash = hashlib.sha256((password + salt).encode()).hexdigest()
# Store user (in production: use database)
USERS_DB[username] = {
'password_hash': password_hash,
'salt': salt,
'email': email,
'role': 'user',
'active': True,
'created_at': datetime.utcnow().isoformat()
}
return jsonify({
'success': True,
'message': 'Registration successful'
}), 201
except Exception as e:
return jsonify({
'success': False,
'error': 'server_error'
}), 500
# Error handlers
@app.errorhandler(404)
def not_found(error):
return jsonify({
'success': False,
'error': 'not_found',
'message': 'Endpoint not found'
}), 404
@app.errorhandler(405)
def method_not_allowed(error):
return jsonify({
'success': False,
'error': 'method_not_allowed',
'message': 'Method not allowed'
}), 405
if __name__ == '__main__':
# Development only - use production WSGI server in production
app.run(debug=True, host='0.0.0.0', port=5000)| Feature | Implementation |
|---|---|
| Rate Limiting | IP-based attempt tracking with lockout |
| Input Sanitization | Removes control characters, validates length |
| Password Hashing | SHA-256 with salt (upgrade to bcrypt/argon2 for production) |
| Secure Cookies | HttpOnly, Secure, SameSite flags |
| Generic Errors | Prevents user enumeration attacks |
| Session Tokens | Cryptographically secure random tokens |
| Security Headers | X-Frame-Options, X-Content-Type-Options, etc. |
# Successful login
curl -X POST http://localhost:5000/login \
-H "Content-Type: application/json" \
-d '{"username": "admin", "password": "admin123"}'
# Failed login (rate limiting test)
curl -X POST http://localhost:5000/login \
-H "Content-Type: application/json" \
-d '{"username": "admin", "password": "wrong"}'For production deployment, replace the in-memory storage with a proper database (PostgreSQL/MySQL), use bcrypt or argon2 for password hashing, and implement JWT or server-side session management with Redis.