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.
47 lines
1.4 KiB
Python
47 lines
1.4 KiB
Python
import jwt
|
|
import re
|
|
from db.db import users_db
|
|
from tinydb import Query
|
|
from flask import request, current_app, jsonify
|
|
|
|
from events.sse import send_event_to_user
|
|
|
|
|
|
def normalize_email(email: str) -> str:
|
|
"""Normalize email for uniqueness checks (Gmail: remove dots and +aliases)."""
|
|
email = email.strip().lower()
|
|
if '@' not in email:
|
|
return email
|
|
local, domain = email.split('@', 1)
|
|
if domain in ('gmail.com', 'googlemail.com'):
|
|
local = local.split('+', 1)[0].replace('.', '')
|
|
return f"{local}@{domain}"
|
|
|
|
def sanitize_email(email):
|
|
return email.replace('@', '_at_').replace('.', '_dot_')
|
|
|
|
def get_current_user_id():
|
|
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')
|
|
if not user_id:
|
|
return None
|
|
return user_id
|
|
except jwt.InvalidTokenError:
|
|
return None
|
|
|
|
def get_validated_user_id():
|
|
user_id = get_current_user_id()
|
|
if not user_id or not users_db.get(Query().id == user_id):
|
|
return None
|
|
return user_id
|
|
|
|
def send_event_for_current_user(event):
|
|
user_id = get_current_user_id()
|
|
if not user_id:
|
|
return jsonify({'error': 'Unauthorized'}), 401
|
|
send_event_to_user(user_id, event)
|
|
return None |