aditya
9fe3196827
All sims re-run after pipeline fixes (better prompts, real syntax validation, fenced-block extraction fix). Results: 7/7 PASS, 20 files total, all Python files pass compile() syntax check. Includes run_all_sims.py runner script and rag-basic action config.
119 lines
3.6 KiB
Python
119 lines
3.6 KiB
Python
import jwt
|
|
from datetime import datetime, timedelta
|
|
from functools import wraps
|
|
from flask import request, jsonify, current_app
|
|
from models import User
|
|
import re
|
|
|
|
def generate_jwt_token(user_id):
|
|
"""Generate JWT token for user."""
|
|
payload = {
|
|
'user_id': user_id,
|
|
'exp': datetime.utcnow() + timedelta(hours=current_app.config['JWT_EXPIRATION_HOURS']),
|
|
'iat': datetime.utcnow()
|
|
}
|
|
|
|
token = jwt.encode(
|
|
payload,
|
|
current_app.config['JWT_SECRET_KEY'],
|
|
algorithm='HS256'
|
|
)
|
|
|
|
return token
|
|
|
|
def verify_jwt_token(token):
|
|
"""Verify JWT token and return user_id if valid."""
|
|
try:
|
|
payload = jwt.decode(
|
|
token,
|
|
current_app.config['JWT_SECRET_KEY'],
|
|
algorithms=['HS256']
|
|
)
|
|
return payload['user_id']
|
|
except jwt.ExpiredSignatureError:
|
|
return None
|
|
except jwt.InvalidTokenError:
|
|
return None
|
|
|
|
def token_required(f):
|
|
"""Decorator to require valid JWT token for protected routes."""
|
|
@wraps(f)
|
|
def decorated(*args, **kwargs):
|
|
token = None
|
|
auth_header = request.headers.get('Authorization')
|
|
|
|
if auth_header and auth_header.startswith('Bearer '):
|
|
token = auth_header.split(' ')[1]
|
|
|
|
if not token:
|
|
return jsonify({
|
|
'error': 'Token is missing',
|
|
'message': 'Authorization token is required'
|
|
}), 401
|
|
|
|
user_id = verify_jwt_token(token)
|
|
if user_id is None:
|
|
return jsonify({
|
|
'error': 'Token is invalid or expired',
|
|
'message': 'Please log in again'
|
|
}), 401
|
|
|
|
current_user = User.query.get(user_id)
|
|
if not current_user or not current_user.is_active:
|
|
return jsonify({
|
|
'error': 'User not found or inactive',
|
|
'message': 'Invalid user credentials'
|
|
}), 401
|
|
|
|
return f(current_user, *args, **kwargs)
|
|
|
|
return decorated
|
|
|
|
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 validate_username(username):
|
|
"""Validate username format."""
|
|
if not username or len(username) < 3 or len(username) > 80:
|
|
return False
|
|
|
|
pattern = r'^[a-zA-Z0-9_]+$'
|
|
return re.match(pattern, username) is not None
|
|
|
|
def validate_password_strength(password):
|
|
"""Validate password strength."""
|
|
min_length = current_app.config.get('PASSWORD_MIN_LENGTH', 8)
|
|
|
|
if not password or len(password) < min_length:
|
|
return False, f'Password must be at least {min_length} characters long'
|
|
|
|
if not re.search(r'[A-Z]', password):
|
|
return False, 'Password must contain at least one uppercase letter'
|
|
|
|
if not re.search(r'[a-z]', password):
|
|
return False, 'Password must contain at least one lowercase letter'
|
|
|
|
if not re.search(r'\d', password):
|
|
return False, 'Password must contain at least one digit'
|
|
|
|
if not re.search(r'[!@#$%^&*(),.?":{}|<>]', password):
|
|
return False, 'Password must contain at least one special character'
|
|
|
|
return True, 'Password is valid'
|
|
|
|
def get_current_user_from_token():
|
|
"""Extract current user from request token."""
|
|
auth_header = request.headers.get('Authorization')
|
|
|
|
if not auth_header or not auth_header.startswith('Bearer '):
|
|
return None
|
|
|
|
token = auth_header.split(' ')[1]
|
|
user_id = verify_jwt_token(token)
|
|
|
|
if user_id:
|
|
return User.query.get(user_id)
|
|
|
|
return None |