Ryan Kegel
6f5b61de7f
All checks were successful
Gitea Actions Demo / build-and-push (push) Successful in 50s
39 lines
1.1 KiB
Python
39 lines
1.1 KiB
Python
import jwt
|
|
import re
|
|
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'])
|
|
email = payload.get('email')
|
|
if not email:
|
|
return None
|
|
return sanitize_email(email)
|
|
except jwt.InvalidTokenError:
|
|
return None
|
|
|
|
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 |