158 lines
No EOL
4.4 KiB
Python
158 lines
No EOL
4.4 KiB
Python
"""Auth and user management API routes"""
|
|
|
|
from flask import Blueprint, current_app, jsonify, request
|
|
from flask_jwt_extended import (
|
|
create_access_token,
|
|
create_refresh_token,
|
|
get_jwt_identity,
|
|
jwt_required,
|
|
)
|
|
|
|
from app import db
|
|
from app.models import User
|
|
from app.services.auth_service import (
|
|
authenticate_user,
|
|
change_password,
|
|
get_login_history,
|
|
register_user,
|
|
)
|
|
|
|
api_bp = Blueprint("api", __name__)
|
|
|
|
|
|
# User Routes
|
|
@api_bp.route("/auth/register", methods=["POST"])
|
|
def register():
|
|
"""Register a new user"""
|
|
if not current_app.config.get("REGISTRATION_ENABLED", True):
|
|
return jsonify({"error": "Not found"}), 404
|
|
|
|
try:
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return jsonify({"error": "Request body is required"}), 400
|
|
|
|
# Server-side validation: require confirm_password for all registrations
|
|
confirm = data.get("confirm_password")
|
|
if not confirm:
|
|
return jsonify({"error": "confirm_password is required"}), 400
|
|
|
|
# Use the auth service for registration with validation
|
|
user = register_user(data)
|
|
|
|
# Generate tokens so user is logged in immediately
|
|
access_token = create_access_token(identity=str(user.id))
|
|
refresh_token = create_refresh_token(identity=str(user.id))
|
|
|
|
return (
|
|
jsonify(
|
|
{
|
|
"user": user.to_dict(),
|
|
"access_token": access_token,
|
|
"refresh_token": refresh_token,
|
|
}
|
|
),
|
|
201,
|
|
)
|
|
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@api_bp.route("/auth/login", methods=["POST"])
|
|
def login():
|
|
"""Login user with account lockout and login history tracking"""
|
|
try:
|
|
data = request.get_json()
|
|
|
|
if not data or not data.get("email") or not data.get("password"):
|
|
return jsonify({"error": "Email and password are required"}), 400
|
|
|
|
# Capture IP and user agent for login history
|
|
ip_address = request.remote_addr or "unknown"
|
|
user_agent = request.headers.get("User-Agent", "unknown")
|
|
|
|
user, access_token, refresh_token = authenticate_user(
|
|
email=data["email"],
|
|
password=data["password"],
|
|
ip_address=ip_address,
|
|
user_agent=user_agent,
|
|
)
|
|
|
|
return (
|
|
jsonify(
|
|
{
|
|
"user": user.to_dict(),
|
|
"access_token": access_token,
|
|
"refresh_token": refresh_token,
|
|
}
|
|
),
|
|
200,
|
|
)
|
|
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 401
|
|
|
|
|
|
@api_bp.route("/users/me", methods=["GET"])
|
|
@jwt_required()
|
|
def get_current_user():
|
|
"""Get current user profile"""
|
|
user_id = int(get_jwt_identity())
|
|
user = db.session.get(User, user_id)
|
|
|
|
if not user:
|
|
return jsonify({"error": "User not found"}), 404
|
|
|
|
return jsonify(user.to_dict()), 200
|
|
|
|
|
|
@api_bp.route("/users/me/password", methods=["PUT"])
|
|
@jwt_required()
|
|
def update_password():
|
|
"""Change current user's password"""
|
|
try:
|
|
data = request.get_json()
|
|
|
|
if not data:
|
|
return jsonify({"error": "Request body is required"}), 400
|
|
|
|
current_password = data.get("current_password")
|
|
new_password = data.get("new_password")
|
|
confirm_password = data.get("confirm_password")
|
|
|
|
if not all([current_password, new_password, confirm_password]):
|
|
return jsonify(
|
|
{
|
|
"error": "current_password, new_password, and confirm_password are required"
|
|
}
|
|
), 400
|
|
|
|
user_id = int(get_jwt_identity())
|
|
user = db.session.get(User, user_id)
|
|
|
|
if not user:
|
|
return jsonify({"error": "User not found"}), 404
|
|
|
|
change_password(user, current_password, new_password, confirm_password)
|
|
|
|
return jsonify({"message": "Password updated successfully"}), 200
|
|
|
|
except ValueError as e:
|
|
return jsonify({"error": str(e)}), 400
|
|
|
|
|
|
@api_bp.route("/users/me/login-history", methods=["GET"])
|
|
@jwt_required()
|
|
def get_user_login_history():
|
|
"""Get login history for current user"""
|
|
user_id = int(get_jwt_identity())
|
|
|
|
limit = request.args.get("limit", 20, type=int)
|
|
# Sanity check on limit
|
|
limit = min(max(limit, 1), 100)
|
|
|
|
history = get_login_history(user_id, limit=limit)
|
|
|
|
return jsonify(history), 200 |