Ryan Kegel
f14de28daa
All checks were successful
Gitea Actions Demo / build-and-push (push) Successful in 36s
- Added `get_validated_user_id` utility function to validate user authentication across multiple APIs. - Updated image upload, request, and listing endpoints to ensure user ownership and proper error handling. - Enhanced reward management endpoints to include user validation and ownership checks. - Modified task management endpoints to enforce user authentication and ownership verification. - Updated models to include `user_id` for images, rewards, tasks, and children to track ownership. - Implemented frontend changes to ensure UI reflects the ownership of tasks and rewards. - Added a new feature specification to prevent deletion of system tasks and rewards.
174 lines
6.5 KiB
Python
174 lines
6.5 KiB
Python
from flask import Blueprint, request, jsonify, current_app
|
|
from models.user import User
|
|
from tinydb import Query
|
|
from db.db import users_db
|
|
import jwt
|
|
import random
|
|
import string
|
|
import smtplib
|
|
from backend.utils.email_instance import email_sender
|
|
from datetime import datetime, timedelta
|
|
from api.utils import get_validated_user_id
|
|
|
|
user_api = Blueprint('user_api', __name__)
|
|
UserQuery = Query()
|
|
|
|
def get_current_user():
|
|
token = request.cookies.get('token')
|
|
if not token:
|
|
return None
|
|
try:
|
|
payload = jwt.decode(token, current_app.config['SECRET_KEY'], algorithms=['HS256'])
|
|
user_id = payload.get('user_id')
|
|
user_dict = users_db.get(UserQuery.id == user_id)
|
|
return User.from_dict(user_dict) if user_dict else None
|
|
except Exception:
|
|
return None
|
|
|
|
@user_api.route('/user/profile', methods=['GET'])
|
|
def get_profile():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
return jsonify({
|
|
'first_name': user.first_name,
|
|
'last_name': user.last_name,
|
|
'email': user.email,
|
|
'image_id': user.image_id
|
|
}), 200
|
|
|
|
@user_api.route('/user/profile', methods=['PUT'])
|
|
def update_profile():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
data = request.get_json()
|
|
# Only allow first_name, last_name, image_id to be updated
|
|
first_name = data.get('first_name')
|
|
last_name = data.get('last_name')
|
|
image_id = data.get('image_id')
|
|
if first_name is not None:
|
|
user.first_name = first_name
|
|
if last_name is not None:
|
|
user.last_name = last_name
|
|
if image_id is not None:
|
|
user.image_id = image_id
|
|
users_db.update(user.to_dict(), UserQuery.email == user.email)
|
|
return jsonify({'message': 'Profile updated'}), 200
|
|
|
|
@user_api.route('/user/image', methods=['PUT'])
|
|
def update_image():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
data = request.get_json()
|
|
image_id = data.get('image_id')
|
|
if not image_id:
|
|
return jsonify({'error': 'Missing image_id'}), 400
|
|
user.image_id = image_id
|
|
users_db.update(user.to_dict(), UserQuery.email == user.email)
|
|
return jsonify({'message': 'Image updated', 'image_id': image_id}), 200
|
|
|
|
@user_api.route('/user/check-pin', methods=['POST'])
|
|
def check_pin():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
data = request.get_json()
|
|
pin = data.get('pin')
|
|
if not pin:
|
|
return jsonify({'error': 'Missing pin'}), 400
|
|
if user.pin and pin == user.pin:
|
|
return jsonify({'valid': True}), 200
|
|
return jsonify({'valid': False}), 200
|
|
|
|
@user_api.route('/user/has-pin', methods=['GET'])
|
|
def has_pin():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
return jsonify({'has_pin': bool(user.pin)}), 200
|
|
|
|
@user_api.route('/user/request-pin-setup', methods=['POST'])
|
|
def request_pin_setup():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user or not user.verified:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
# Generate 6-digit/character code
|
|
code = ''.join(random.choices(string.ascii_uppercase + string.digits, k=6))
|
|
user.pin_setup_code = code
|
|
user.pin_setup_code_created = datetime.utcnow().isoformat()
|
|
users_db.update(user.to_dict(), UserQuery.email == user.email)
|
|
# Send email
|
|
send_pin_setup_email(user.email, code)
|
|
return jsonify({'message': 'Verification code sent to your email.'}), 200
|
|
|
|
def send_pin_setup_email(email, code):
|
|
# Use the reusable email sender
|
|
email_sender.send_pin_setup_email(email, code)
|
|
|
|
@user_api.route('/user/verify-pin-setup', methods=['POST'])
|
|
def verify_pin_setup():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user or not user.verified:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
data = request.get_json()
|
|
code = data.get('code')
|
|
if not code:
|
|
return jsonify({'error': 'Missing code'}), 400
|
|
if not user.pin_setup_code or not user.pin_setup_code_created:
|
|
return jsonify({'error': 'No code requested'}), 400
|
|
# Check expiry (10 min)
|
|
created = datetime.fromisoformat(user.pin_setup_code_created)
|
|
if datetime.utcnow() > created + timedelta(minutes=10):
|
|
return jsonify({'error': 'Code expired'}), 400
|
|
if code.strip().upper() != user.pin_setup_code.upper():
|
|
return jsonify({'error': 'Invalid code'}), 400
|
|
return jsonify({'message': 'Code verified'}), 200
|
|
|
|
@user_api.route('/user/set-pin', methods=['POST'])
|
|
def set_pin():
|
|
user_id = get_validated_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
|
user = get_current_user()
|
|
if not user or not user.verified:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
data = request.get_json()
|
|
pin = data.get('pin')
|
|
if not pin or not pin.isdigit() or not (4 <= len(pin) <= 6):
|
|
return jsonify({'error': 'PIN must be 4-6 digits'}), 400
|
|
# Only allow if code was recently verified
|
|
if not user.pin_setup_code or not user.pin_setup_code_created:
|
|
return jsonify({'error': 'No code verified'}), 400
|
|
created = datetime.fromisoformat(user.pin_setup_code_created)
|
|
if datetime.utcnow() > created + timedelta(minutes=10):
|
|
return jsonify({'error': 'Code expired'}), 400
|
|
# Set pin, clear code
|
|
user.pin = pin
|
|
user.pin_setup_code = ''
|
|
user.pin_setup_code_created = None
|
|
users_db.update(user.to_dict(), UserQuery.email == user.email)
|
|
return jsonify({'message': 'Parent PIN set'}), 200
|