feat: Implement user validation and ownership checks for image, reward, and task APIs
All checks were successful
Gitea Actions Demo / build-and-push (push) Successful in 36s
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.
This commit is contained in:
@@ -4,7 +4,7 @@ from PIL import Image as PILImage, UnidentifiedImageError
|
||||
from flask import Blueprint, request, jsonify, send_file
|
||||
from tinydb import Query
|
||||
|
||||
from api.utils import get_current_user_id, sanitize_email
|
||||
from api.utils import get_current_user_id, sanitize_email, get_validated_user_id
|
||||
from config.paths import get_user_image_dir
|
||||
|
||||
from db.db import image_db
|
||||
@@ -21,9 +21,9 @@ def allowed_file(filename):
|
||||
|
||||
@image_api.route('/image/upload', methods=['POST'])
|
||||
def upload():
|
||||
user_id = get_current_user_id()
|
||||
user_id = get_validated_user_id()
|
||||
if not user_id:
|
||||
return jsonify({'error': 'User not authenticated'}), 401
|
||||
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
||||
if 'file' not in request.files:
|
||||
return jsonify({'error': 'No file part in the request'}), 400
|
||||
file = request.files['file']
|
||||
@@ -64,8 +64,11 @@ def upload():
|
||||
|
||||
format_extension_map = {'JPEG': '.jpg', 'PNG': '.png'}
|
||||
extension = format_extension_map.get(original_format, '.png')
|
||||
image_record = Image(extension=extension, permanent=perm, type=image_type, user=user_id)
|
||||
image_record = Image(extension=extension, permanent=perm, type=image_type, user_id=user_id)
|
||||
filename = image_record.id + extension
|
||||
user_image_dir = get_user_image_dir(user_id)
|
||||
os.makedirs(user_image_dir, exist_ok=True)
|
||||
|
||||
filepath = os.path.abspath(os.path.join(get_user_image_dir(sanitize_email(user_id)), filename))
|
||||
|
||||
try:
|
||||
@@ -84,25 +87,35 @@ def upload():
|
||||
|
||||
@image_api.route('/image/request/<id>', methods=['GET'])
|
||||
def request_image(id):
|
||||
user_id = get_validated_user_id()
|
||||
if not user_id:
|
||||
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
||||
ImageQuery = Query()
|
||||
image: Image = Image.from_dict(image_db.get(ImageQuery.id == id))
|
||||
if not image:
|
||||
image_record = image_db.get(ImageQuery.id == id)
|
||||
if not image_record:
|
||||
return jsonify({'error': 'Image not found'}), 404
|
||||
image = Image.from_dict(image_record)
|
||||
# Allow if image.user_id is None (public image), or matches user_id
|
||||
if image.user_id is not None and image.user_id != user_id:
|
||||
return jsonify({'error': 'Forbidden: image does not belong to user', 'code': 'FORBIDDEN'}), 403
|
||||
filename = f"{image.id}{image.extension}"
|
||||
filepath = os.path.abspath(os.path.join(get_user_image_dir(image.user), filename))
|
||||
filepath = os.path.abspath(os.path.join(get_user_image_dir(image.user_id or user_id), filename))
|
||||
if not os.path.exists(filepath):
|
||||
return jsonify({'error': 'File not found'}), 404
|
||||
return send_file(filepath)
|
||||
|
||||
@image_api.route('/image/list', methods=['GET'])
|
||||
def list_images():
|
||||
user_id = get_validated_user_id()
|
||||
if not user_id:
|
||||
return jsonify({'error': 'Unauthorized', 'code': 'UNAUTHORIZED'}), 401
|
||||
image_type = request.args.get('type', type=int)
|
||||
ImageQuery = Query()
|
||||
if image_type is not None:
|
||||
if image_type not in [IMAGE_TYPE_PROFILE, IMAGE_TYPE_ICON]:
|
||||
return jsonify({'error': 'Invalid image type'}), 400
|
||||
images = image_db.search(ImageQuery.type == image_type)
|
||||
images = image_db.search((ImageQuery.type == image_type) & ((ImageQuery.user_id == user_id) | (ImageQuery.user_id == None)))
|
||||
else:
|
||||
images = image_db.all()
|
||||
images = image_db.search((ImageQuery.user_id == user_id) | (ImageQuery.user_id == None))
|
||||
image_ids = [img['id'] for img in images]
|
||||
return jsonify({'ids': image_ids, 'count': len(image_ids)}), 200
|
||||
|
||||
Reference in New Issue
Block a user