Implement Firebase push notifications

This commit is contained in:
2025-12-29 12:44:29 +03:00
Unverified
parent b31d31c2d1
commit a325de8f39
7 changed files with 288 additions and 27 deletions
+10
View File
@@ -108,6 +108,16 @@ app.add_middleware(SlowAPIMiddleware)
@app.middleware("http") @app.middleware("http")
async def access_logging_middleware(request: Request, call_next): async def access_logging_middleware(request: Request, call_next):
# Log incoming request and Authorization header presence for debugging auth issues
try:
auth_header = request.headers.get("authorization")
if auth_header:
short = auth_header[:20] + "..." if len(auth_header) > 20 else auth_header
logger.info("Incoming request %s %s Authorization=%s", request.method, request.url.path, short)
else:
logger.info("Incoming request %s %s Authorization=NONE", request.method, request.url.path)
except Exception:
pass
start = time.perf_counter() start = time.perf_counter()
try: try:
response = await call_next(request) response = await call_next(request)
+17 -1
View File
@@ -5,8 +5,10 @@ from sqlalchemy.orm import Session
from utils import verify_token from utils import verify_token
from models import User, DeviceSession from models import User, DeviceSession
from db import SessionLocal from db import SessionLocal
import logging
security = HTTPBearer() security = HTTPBearer()
logger = logging.getLogger("uvicorn.error")
# Зависимость для получения сессии БД # Зависимость для получения сессии БД
def get_db(): def get_db():
@@ -23,8 +25,17 @@ def get_current_user(
db: Session = Depends(get_db), db: Session = Depends(get_db),
) -> User: ) -> User:
token = credentials.credentials token = credentials.credentials
payload = verify_token(token) try:
payload = verify_token(token)
except Exception as e:
logger.warning("get_current_user: token verification error: %s", str(e))
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired token",
headers={"WWW-Authenticate": "Bearer"},
)
if not payload: if not payload:
logger.info("get_current_user: verify_token returned empty payload")
raise HTTPException( raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, status_code=status.HTTP_401_UNAUTHORIZED,
detail="Invalid or expired token", detail="Invalid or expired token",
@@ -32,6 +43,7 @@ def get_current_user(
) )
user = db.query(User).filter(User.id == payload["user_id"]).first() user = db.query(User).filter(User.id == payload["user_id"]).first()
if not user: if not user:
logger.info("get_current_user: user not found for user_id=%s", payload.get("user_id"))
raise HTTPException( raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, status_code=status.HTTP_401_UNAUTHORIZED,
detail="User not found", detail="User not found",
@@ -60,6 +72,7 @@ def get_current_user(
) )
if not device_session or device_session.revoked: if not device_session or device_session.revoked:
logger.info("get_current_user: session missing/revoked for user_id=%s session_id=%s", user.id, session_id)
raise HTTPException( raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, status_code=status.HTTP_401_UNAUTHORIZED,
detail="Session revoked or not found", detail="Session revoked or not found",
@@ -73,6 +86,7 @@ def get_current_user(
# Session expired due to inactivity - revoke it # Session expired due to inactivity - revoke it
device_session.revoked = True device_session.revoked = True
db.commit() db.commit()
logger.info("get_current_user: session expired due to inactivity for user_id=%s session_id=%s", user.id, session_id)
raise HTTPException( raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED, status_code=status.HTTP_401_UNAUTHORIZED,
detail="Session expired due to inactivity", detail="Session expired due to inactivity",
@@ -85,6 +99,7 @@ def get_current_user(
# Check if user is suspended # Check if user is suspended
if user.suspended: if user.suspended:
logger.info("get_current_user: account suspended for user_id=%s reason=%s", user.id, user.suspension_reason)
raise HTTPException( raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, status_code=status.HTTP_403_FORBIDDEN,
detail="Account suspended", detail="Account suspended",
@@ -93,6 +108,7 @@ def get_current_user(
# Check if user is deleted # Check if user is deleted
if user.deleted: if user.deleted:
logger.info("get_current_user: account deleted for user_id=%s", user.id)
raise HTTPException( raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN, status_code=status.HTTP_403_FORBIDDEN,
detail="Account deleted", detail="Account deleted",
+10
View File
@@ -113,6 +113,16 @@ class PushSubscription(Base):
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now) updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
class FcmToken(Base):
__tablename__ = "fcm_token"
id = Column(Integer, primary_key=True, index=True)
user_id = Column(Integer, ForeignKey("user.id"), nullable=False, index=True)
token = Column(Text, nullable=False, unique=True)
created_at = Column(DateTime, default=datetime.now)
updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)
class Reaction(Base): class Reaction(Base):
__tablename__ = "reaction" __tablename__ = "reaction"
+118 -26
View File
@@ -5,6 +5,11 @@ from typing import List, Optional
from sqlalchemy.orm import Session from sqlalchemy.orm import Session
from pywebpush import webpush, WebPushException from pywebpush import webpush, WebPushException
from models import PushSubscription, User, Message, DMEnvelope from models import PushSubscription, User, Message, DMEnvelope
from models import FcmToken
import firebase_admin
from firebase_admin import credentials as firebase_credentials
from firebase_admin import messaging as firebase_messaging
import base64
logger = logging.getLogger("uvicorn.error") logger = logging.getLogger("uvicorn.error")
@@ -12,6 +17,24 @@ class PushNotificationService:
def __init__(self): def __init__(self):
self.vapid_private_key = os.getenv("VAPID_PRIVATE_KEY") self.vapid_private_key = os.getenv("VAPID_PRIVATE_KEY")
self.vapid_public_key = os.getenv("VAPID_PUBLIC_KEY") self.vapid_public_key = os.getenv("VAPID_PUBLIC_KEY")
# Firebase Admin initialization (modern API). Only FIREBASE_CERT env is supported.
self.firebase_initialized = False
try:
firebase_cert = os.getenv("FIREBASE_CERT")
if not firebase_cert:
raise RuntimeError("FIREBASE_CERT env variable is required for Firebase Admin SDK initialization")
# Support raw JSON or base64-encoded JSON in FIREBASE_CERT
decoded = base64.b64decode(firebase_cert).decode("utf-8")
sa_dict = json.loads(decoded)
cred = firebase_credentials.Certificate(sa_dict)
firebase_admin.initialize_app(cred)
self.firebase_initialized = True
logger.info("Firebase Admin SDK initialized for push sending (FIREBASE_CERT)")
except Exception as e:
logger.error(f"Failed to initialize Firebase Admin SDK from FIREBASE_CERT: {e}")
raise
if (not self.vapid_public_key) or (not self.vapid_private_key): if (not self.vapid_public_key) or (not self.vapid_private_key):
raise ValueError("VAPID public or private key is None") raise ValueError("VAPID public or private key is None")
@@ -57,42 +80,61 @@ class PushNotificationService:
users = db.query(User).filter(User.id != message.user_id) users = db.query(User).filter(User.id != message.user_id)
if exclude_user_id: if exclude_user_id:
users = users.filter(User.id != exclude_user_id) users = users.filter(User.id != exclude_user_id)
for user in users: for user in users:
# Check if user has push subscription before trying to send # Check if user has push subscription before trying to send
# Try all FCM tokens first (Android). If none or all fail, fall back to web push subscription.
fcm_rows = db.query(FcmToken).filter(FcmToken.user_id == user.id).all()
payload_data = {
"type": "public_message",
"message_id": message.id,
"sender_id": message.user_id,
"sender_username": message.author.username
}
title = f"{message.author.username}"
body = message.content[:100] + ("..." if len(message.content) > 100 else "")
if fcm_rows and self.firebase_initialized:
for fcm in fcm_rows:
try:
self._send_fcm_to_token(fcm.token, title, body, payload_data)
except Exception as e:
logger.error(f"Failed to send FCM to user {user.id} token {fcm.token}: {e}")
# Check if this is a permanent failure and clean up the token
self._cleanup_failed_fcm_token(db, fcm, str(e))
subscription = db.query(PushSubscription).filter(PushSubscription.user_id == user.id).first() subscription = db.query(PushSubscription).filter(PushSubscription.user_id == user.id).first()
if not subscription: if subscription:
continue await self._send_notification_to_user(
db, user.id, title, body, message.author.profile_picture, payload_data
await self._send_notification_to_user( )
db, user.id,
f"New message from {message.author.username}",
message.content[:100] + ("..." if len(message.content) > 100 else ""),
message.author.profile_picture,
{
"type": "public_message",
"message_id": message.id,
"sender_id": message.user_id,
"sender_username": message.author.username
}
)
except Exception as e: except Exception as e:
logger.error(f"Failed to send public message notifications: {e}") logger.error(f"Failed to send public message notifications: {e}")
async def send_dm_notification(self, db: Session, dm_envelope: DMEnvelope, sender: User): async def send_dm_notification(self, db: Session, dm_envelope: DMEnvelope, sender: User):
"""Send push notification for a new DM""" """Send push notification for a new DM"""
try: try:
title = f"{sender.username}"
body = "New direct message"
payload_data = {
"type": "dm",
"dm_id": dm_envelope.id,
"sender_id": sender.id,
"sender_username": sender.username
}
fcm_rows = db.query(FcmToken).filter(FcmToken.user_id == dm_envelope.recipient_id).all()
if fcm_rows and self.firebase_initialized:
for fcm in fcm_rows:
try:
self._send_fcm_to_token(fcm.token, title, body, payload_data)
except Exception as e:
logger.error(f"Failed to send FCM to user {dm_envelope.recipient_id} token {fcm.token}: {e}")
# Check if this is a permanent failure and clean up the token
self._cleanup_failed_fcm_token(db, fcm, str(e))
await self._send_notification_to_user( await self._send_notification_to_user(
db, dm_envelope.recipient_id, db, dm_envelope.recipient_id, title, body, sender.profile_picture, payload_data
f"New message from {sender.username}",
"You have a new direct message",
sender.profile_picture,
{
"type": "dm",
"dm_id": dm_envelope.id,
"sender_id": sender.id,
"sender_username": sender.username
}
) )
except Exception as e: except Exception as e:
logger.error(f"Failed to send DM notification: {e}") logger.error(f"Failed to send DM notification: {e}")
@@ -136,6 +178,56 @@ class PushNotificationService:
except Exception as e: except Exception as e:
logger.error(f"Failed to send push notification to user {user_id}: {e}") logger.error(f"Failed to send push notification to user {user_id}: {e}")
def _send_fcm_to_token(self, token: str, title: str, body: str, data: dict):
"""Send an FCM data-only push to a single device token using Firebase Admin SDK.
Notification display is handled by the app, not FCM."""
if not self.firebase_initialized:
raise RuntimeError("Firebase Admin SDK not initialized (FIREBASE_CERT required)")
try:
# Send only data payload - let the app handle notification display
# This prevents FCM from auto-showing notifications
msg = firebase_messaging.Message(
token=token,
data={
"title": title,
"body": body,
**{k: str(v) for k, v in (data or {}).items()}
},
android=firebase_messaging.AndroidConfig(priority="high"),
apns=firebase_messaging.APNSConfig(headers={"apns-priority": "10"})
)
resp = firebase_messaging.send(msg)
return resp
except Exception as e:
logger.error(f"Firebase Admin send failed for token {token}: {e}")
raise
def _cleanup_failed_fcm_token(self, db: Session, fcm_token_entry, error_message: str):
"""Clean up FCM tokens that have permanent failures"""
try:
# Check for permanent failure indicators in the error message
permanent_errors = [
"unregistered", "invalidregistration", "notregistered",
"sender_id_mismatch", "invalid_argument"
]
error_lower = error_message.lower()
is_permanent = any(permanent_error in error_lower for permanent_error in permanent_errors)
if is_permanent:
logger.info(f"Removing permanently failed FCM token for user {fcm_token_entry.user_id}: {fcm_token_entry.token}")
db.query(FcmToken).filter(FcmToken.id == fcm_token_entry.id).delete()
db.commit()
else:
logger.debug(f"Temporary FCM failure for token {fcm_token_entry.token}, keeping token: {error_message}")
except Exception as e:
logger.error(f"Failed to cleanup FCM token {fcm_token_entry.token}: {e}")
try:
db.rollback()
except Exception:
pass
async def unsubscribe_user(self, db: Session, user_id: int) -> bool: async def unsubscribe_user(self, db: Session, user_id: int) -> bool:
"""Unsubscribe a user from push notifications""" """Unsubscribe a user from push notifications"""
try: try:
+1
View File
@@ -14,3 +14,4 @@ user-agents>=2.2.0
httpx>=0.27.2 httpx>=0.27.2
rich>=13.9.4 rich>=13.9.4
slowapi>=0.1.9 slowapi>=0.1.9
firebase_admin>=7.1.0
+131
View File
@@ -24,12 +24,15 @@ from push_service import push_service
from PIL import Image from PIL import Image
import io import io
import json import json
from pydantic import BaseModel
from better_profanity import profanity as _bp from better_profanity import profanity as _bp
from security.audit import log_access, log_dm, log_public_chat, log_security from security.audit import log_access, log_dm, log_public_chat, log_security
from security.profanity import contains_profanity from security.profanity import contains_profanity
from security.rate_limit import rate_limit_per_ip from security.rate_limit import rate_limit_per_ip
from websocket.utils import authenticate_user from websocket.utils import authenticate_user
from models import FcmToken
router = APIRouter() router = APIRouter()
logger = logging.getLogger("uvicorn.error") logger = logging.getLogger("uvicorn.error")
@@ -452,6 +455,97 @@ async def send_message(
return await _send_message_internal(message_request, current_user, db, files) return await _send_message_internal(message_request, current_user, db, files)
class RegisterFcmRequest(BaseModel):
token: str
@router.post("/push/register")
async def register_fcm_token(request: Request, body: RegisterFcmRequest, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
"""
Register or update an FCM token for the authenticated user.
"""
token = body.token.strip() if body and body.token else None
if not token:
raise HTTPException(status_code=400, detail="Missing token")
try:
# If token already exists (from another device), reassign it to this user.
token_row = db.query(FcmToken).filter(FcmToken.token == token).first()
if token_row:
token_row.user_id = current_user.id
else:
# Create new token record (allow multiple tokens per user)
new = FcmToken(user_id=current_user.id, token=token)
db.add(new)
db.commit()
logger.info(f"Registered FCM token for user {current_user.id}: {token}")
except Exception as e:
try:
db.rollback()
except Exception:
pass
raise HTTPException(status_code=500, detail="Failed to save token")
return {"status": "success"}
@router.post("/push/unregister")
async def unregister_fcm_token(request: Request, body: RegisterFcmRequest | None = None, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
"""
Unregister an FCM token. If `body.token` provided, remove only that token for the user.
If no token provided, remove all tokens for the user.
"""
try:
if body and body.token:
db.query(FcmToken).filter(FcmToken.user_id == current_user.id, FcmToken.token == body.token.strip()).delete()
else:
db.query(FcmToken).filter(FcmToken.user_id == current_user.id).delete()
db.commit()
except Exception as e:
try:
db.rollback()
except Exception:
pass
raise HTTPException(status_code=500, detail="Failed to remove token")
return {"status": "success"}
@router.post("/push/test")
async def push_test(request: Request, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
"""
Send a test push to the current user's registered FCM token (for manual testing).
"""
try:
fcm_rows = db.query(FcmToken).filter(FcmToken.user_id == current_user.id).all()
if not fcm_rows:
raise HTTPException(status_code=404, detail="No FCM token registered for user")
title = "FromChat test"
body = "This is a test push from the server"
data = {"type": "test", "timestamp": datetime.utcnow().isoformat()}
# Use push_service which uses Admin SDK internally; attempt to send to all tokens
failures = []
for fcm in fcm_rows:
try:
push_service._send_fcm_to_token(fcm.token, title, body, data)
except Exception as e:
logger.error(f"Failed to send test push to user {current_user.id} token {fcm.token}: {e}")
failures.append(str(e))
if failures and len(failures) == len(fcm_rows):
# All failed
raise HTTPException(status_code=500, detail=f"Failed to send push to any token: {failures}")
return {"status": "success", "sent": len(fcm_rows) - len(failures), "failed": len(failures)}
except HTTPException:
raise
except Exception as e:
logger.error(f"push_test error: {e}")
raise HTTPException(status_code=500, detail="Internal error")
@router.get("/get_messages") @router.get("/get_messages")
@rate_limit_per_ip("60/minute") # Per-IP limit to prevent abuse @rate_limit_per_ip("60/minute") # Per-IP limit to prevent abuse
async def get_messages(request: Request, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)): async def get_messages(request: Request, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
@@ -467,6 +561,43 @@ async def get_messages(request: Request, current_user: User = Depends(get_curren
} }
class MarkReadRequest(BaseModel):
messageIds: list[int]
@router.get("/messages/new")
@rate_limit_per_ip("60/minute")
async def get_new_messages(request: Request, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
"""
Return unread public messages (Message.is_read == False).
"""
new_messages = db.query(Message).filter(Message.is_read == False).order_by(Message.timestamp.asc()).all()
messages_data = [convert_message(msg) for msg in new_messages]
return {"status": "success", "messages": messages_data}
@router.post("/messages/read")
@rate_limit_per_ip("60/minute")
async def mark_messages_read(request: Request, read_request: MarkReadRequest, current_user: User = Depends(get_current_user), db: Session = Depends(get_db)):
"""
Mark specified message IDs as read (set Message.is_read = True).
"""
if not read_request or not isinstance(read_request.messageIds, list) or len(read_request.messageIds) == 0:
return {"status": "success", "updated": 0}
try:
updated_count = db.query(Message).filter(Message.id.in_(read_request.messageIds)).update({Message.is_read: True}, synchronize_session=False)
db.commit()
except Exception as e:
try:
db.rollback()
except Exception:
pass
raise HTTPException(status_code=500, detail="Failed to mark messages as read")
return {"status": "success", "updated": int(updated_count)}
@router.post("/dm/send") @router.post("/dm/send")
@rate_limit_per_ip("20/minute") @rate_limit_per_ip("20/minute")
async def dm_send( async def dm_send(
+1
View File
@@ -9,4 +9,5 @@ JWT_SECRET="$(openssl rand -base64 32)"
TURN_USERNAME=<set> TURN_USERNAME=<set>
TURN_SECRET=<set> TURN_SECRET=<set>
DEPLOYMENT_SERVER=<set> DEPLOYMENT_SERVER=<set>
FIREBASE_CERT=<set>
EOF EOF