Ryan Kegel
7e7a2ef49e
Some checks failed
Chore App Build and Push Docker Images / build-and-push (push) Has been cancelled
- Added checks for accounts marked for deletion in signup, verification, and password reset processes. - Updated reward and task listing to sort user-created items first. - Enhanced user API to clear verification and reset tokens when marking accounts for deletion. - Introduced tests for marked accounts to ensure proper handling in various scenarios. - Updated profile and reward edit components to reflect changes in validation and data handling.
247 lines
9.3 KiB
Python
247 lines
9.3 KiB
Python
from flask import Blueprint, request, jsonify, current_app
|
|
from events.types.user_modified import UserModified
|
|
from models.user import User
|
|
from tinydb import Query
|
|
from db.db import users_db
|
|
import jwt
|
|
import random
|
|
import string
|
|
import utils.email_sender as email_sender
|
|
from datetime import datetime, timedelta, timezone
|
|
from api.utils import get_validated_user_id, normalize_email, send_event_for_current_user
|
|
from api.error_codes import ACCOUNT_MARKED_FOR_DELETION, ALREADY_MARKED
|
|
from events.types.event_types import EventType
|
|
from events.types.event import Event
|
|
from events.types.profile_updated import ProfileUpdated
|
|
from utils.tracking_logger import log_tracking_event
|
|
from models.tracking_event import TrackingEvent
|
|
from db.tracking import insert_tracking_event
|
|
|
|
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)
|
|
|
|
# Create tracking event
|
|
metadata = {}
|
|
if first_name is not None:
|
|
metadata['first_name_updated'] = True
|
|
if last_name is not None:
|
|
metadata['last_name_updated'] = True
|
|
if image_id is not None:
|
|
metadata['image_updated'] = True
|
|
|
|
tracking_event = TrackingEvent.create_event(
|
|
user_id=user_id,
|
|
child_id=None, # No child for user profile
|
|
entity_type='user',
|
|
entity_id=user.id,
|
|
action='updated',
|
|
points_before=0, # Not relevant
|
|
points_after=0,
|
|
metadata=metadata
|
|
)
|
|
insert_tracking_event(tracking_event)
|
|
log_tracking_event(tracking_event)
|
|
|
|
# Send SSE event
|
|
send_event_for_current_user(Event(EventType.PROFILE_UPDATED.value, ProfileUpdated(user.id)))
|
|
|
|
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
|
|
|
|
@user_api.route('/user/mark-for-deletion', methods=['POST'])
|
|
def mark_for_deletion():
|
|
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
|
|
|
|
# Validate email from request body
|
|
data = request.get_json()
|
|
email = data.get('email', '').strip()
|
|
if not email:
|
|
return jsonify({'error': 'Email is required', 'code': 'EMAIL_REQUIRED'}), 400
|
|
|
|
# Verify email matches the logged-in user - make sure to normalize the email address first
|
|
if normalize_email(email) != normalize_email(user.email):
|
|
return jsonify({'error': 'Email does not match your account', 'code': 'EMAIL_MISMATCH'}), 400
|
|
|
|
# Check if already marked
|
|
if user.marked_for_deletion:
|
|
return jsonify({'error': 'Account already marked for deletion', 'code': ALREADY_MARKED}), 400
|
|
|
|
# Mark for deletion
|
|
user.marked_for_deletion = True
|
|
user.marked_for_deletion_at = datetime.now(timezone.utc).isoformat()
|
|
|
|
# Invalidate any outstanding verification/reset tokens so they cannot be used after marking
|
|
user.verify_token = None
|
|
user.verify_token_created = None
|
|
user.reset_token = None
|
|
user.reset_token_created = None
|
|
|
|
users_db.update(user.to_dict(), UserQuery.id == user.id)
|
|
|
|
# Trigger SSE event
|
|
send_event_for_current_user(Event(EventType.USER_MARKED_FOR_DELETION.value, UserModified(user.id, UserModified.OPERATION_DELETE)))
|
|
|
|
return jsonify({'success': True}), 200
|