Implement JWT authentication and user registration/login endpoints in Planventure API

Author: LadyKerr

planventure-api/app.py

@@ -1,8 +1,10 @@
 from flask import Flask, jsonify
 from flask_cors import CORS
 from flask_sqlalchemy import SQLAlchemy
+from flask_jwt_extended import JWTManager
 from os import environ
 from dotenv import load_dotenv
+from datetime import timedelta
 
 # Load environment variables
 load_dotenv()
@@ -14,12 +16,44 @@ def create_app():
     app = Flask(__name__)
     CORS(app)
 
+    # JWT Configuration
+    app.config['JWT_SECRET_KEY'] = environ.get('JWT_SECRET_KEY', 'your-secret-key')
+    app.config['JWT_ACCESS_TOKEN_EXPIRES'] = timedelta(hours=1)
+    jwt = JWTManager(app)
+
+    @jwt.expired_token_loader
+    def expired_token_callback(jwt_header, jwt_data):
+        return jsonify({
+            'error': 'Token has expired',
+            'code': 'token_expired'
+        }), 401
+
+    @jwt.invalid_token_loader
+    def invalid_token_callback(error):
+        return jsonify({
+            'error': 'Invalid token',
+            'code': 'invalid_token'
+        }), 401
+
+    @jwt.unauthorized_loader
+    def missing_token_callback(error):
+        return jsonify({
+            'error': 'Authorization token is missing',
+            'code': 'authorization_required'
+        }), 401
+
     # Database configuration
     app.config['SQLALCHEMY_DATABASE_URI'] = environ.get('DATABASE_URL', 'sqlite:///planventure.db')
     app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
 
     # Initialize extensions
     db.init_app(app)
+    
+    # Register blueprints
+    from routes.auth import auth_bp
+    from routes.trips import trips_bp
+    app.register_blueprint(auth_bp, url_prefix='/auth')
+    app.register_blueprint(trips_bp, url_prefix='/api')
 
     # Register routes
     @app.route('/')

planventure-api/instance/planventure.db

@@ -0,0 +1,21 @@
+from functools import wraps
+from flask import jsonify
+from flask_jwt_extended import verify_jwt_in_request, get_jwt_identity
+from models import User
+
+def auth_middleware(f):
+    @wraps(f)
+    def decorated(*args, **kwargs):
+        try:
+            verify_jwt_in_request(optional=True)
+            current_user_id = get_jwt_identity()
+            
+            # Check if user still exists in database
+            user = User.query.get(current_user_id)
+            if not user:
+                return jsonify({"error": "User not found"}), 401
+                
+            return f(*args, **kwargs)
+        except Exception as e:
+            return jsonify({"error": "Invalid or expired token"}), 401
+    return decorated

planventure-api/middleware/auth.py

@@ -1,5 +1,7 @@
 from datetime import datetime, timezone
+from flask_jwt_extended import create_access_token
 from app import db
+from utils.password import hash_password, check_password
 
 class User(db.Model):
     __tablename__ = 'users'
@@ -13,5 +15,25 @@ class User(db.Model):
     # Add relationship
     trips = db.relationship('Trip', back_populates='user', cascade='all, delete-orphan')
 
+    @property
+    def password(self):
+        raise AttributeError('password is not a readable attribute')
+
+    @password.setter
+    def password(self, password):
+        self.password_hash = hash_password(password)
+
+    def verify_password(self, password):
+        return check_password(password, self.password_hash)
+
+    def generate_auth_token(self):
+        """Generate JWT token for the user"""
+        return create_access_token(identity=self.id)
+
+    @staticmethod
+    def verify_auth_token(token):
+        """Verify the auth token - handled by @auth_required decorator"""
+        pass
+
     def __repr__(self):
         return f'<User {self.email}>'

planventure-api/models/user.py

@@ -0,0 +1,3 @@
+from .auth import auth_bp
+
+__all__ = ['auth_bp']

planventure-api/routes/__init__.py

@@ -0,0 +1,70 @@
+from flask import Blueprint, request, jsonify
+from app import db
+from models import User
+from utils.validators import validate_email
+
+auth_bp = Blueprint('auth', __name__)
+
+# Error responses
+INVALID_CREDENTIALS = {"error": "Invalid email or password"}, 401
+MISSING_FIELDS = {"error": "Missing required fields"}, 400
+INVALID_EMAIL = {"error": "Invalid email format"}, 400
+EMAIL_EXISTS = {"error": "Email already registered"}, 409
+
+@auth_bp.route('/register', methods=['POST'])
+def register():
+    data = request.get_json()
+    
+    # Validate required fields
+    if not all(k in data for k in ['email', 'password']):
+        return jsonify(MISSING_FIELDS)
+    
+    # Validate email format
+    if not validate_email(data['email']):
+        return jsonify(INVALID_EMAIL)
+    
+    # Check if user already exists
+    if User.query.filter_by(email=data['email']).first():
+        return jsonify(EMAIL_EXISTS)
+    
+    # Create new user
+    try:
+        user = User(email=data['email'])
+        user.password = data['password']  # This will hash the password
+        db.session.add(user)
+        db.session.commit()
+        
+        # Generate auth token
+        token = user.generate_auth_token()
+        return jsonify({
+            'message': 'User registered successfully',
+            'token': token
+        }), 201
+    except Exception as e:
+        db.session.rollback()
+        return jsonify({'error': 'Registration failed'}), 500
+
+@auth_bp.route('/login', methods=['POST'])
+def login():
+    data = request.get_json()
+    
+    # Validate required fields
+    if not all(k in data for k in ['email', 'password']):
+        return jsonify(MISSING_FIELDS)
+    
+    # Find user by email
+    user = User.query.filter_by(email=data['email']).first()
+    
+    # Verify user exists and password is correct
+    if user and user.verify_password(data['password']):
+        token = user.generate_auth_token()
+        return jsonify({
+            'message': 'Login successful',
+            'token': token,
+            'user': {
+                'id': user.id,
+                'email': user.email
+            }
+        }), 200
+    
+    return jsonify(INVALID_CREDENTIALS)

planventure-api/routes/auth.py

@@ -0,0 +1,10 @@
+from flask import Blueprint, jsonify
+from middleware.auth import auth_middleware
+
+trips_bp = Blueprint('trips', __name__)
+
+@trips_bp.route('/trips', methods=['GET'])
+@auth_middleware
+def get_trips():
+    # This route is now protected and will only be accessible with a valid JWT token
+    return jsonify({"message": "Protected route accessed successfully"})

planventure-api/routes/trips.py

@@ -0,0 +1,4 @@
+from .password import hash_password, check_password
+from .auth import auth_required, get_current_user_id
+
+__all__ = ['hash_password', 'check_password', 'auth_required', 'get_current_user_id']

planventure-api/utils/__init__.py

@@ -0,0 +1,16 @@
+from functools import wraps
+from flask import jsonify
+from flask_jwt_extended import verify_jwt_in_request, get_jwt_identity
+
+def auth_required(f):
+    @wraps(f)
+    def decorated(*args, **kwargs):
+        try:
+            verify_jwt_in_request()
+            return f(*args, **kwargs)
+        except Exception as e:
+            return jsonify({"msg": "Invalid token"}), 401
+    return decorated
+
+def get_current_user_id():
+    return get_jwt_identity()

planventure-api/utils/auth.py

@@ -0,0 +1,13 @@
+import bcrypt
+
+def hash_password(password: str) -> str:
+    """Hash a password using bcrypt"""
+    salt = bcrypt.gensalt()
+    return bcrypt.hashpw(password.encode('utf-8'), salt).decode('utf-8')
+
+def check_password(password: str, password_hash: str) -> bool:
+    """Verify a password against its hash"""
+    return bcrypt.checkpw(
+        password.encode('utf-8'),
+        password_hash.encode('utf-8')
+    )