361 lines
12 KiB
Python
361 lines
12 KiB
Python
import base64
|
|
import json
|
|
import logging
|
|
import os
|
|
import struct
|
|
import time
|
|
from threading import Thread
|
|
from urllib.parse import urlparse
|
|
|
|
from cryptography.hazmat.primitives import hashes, serialization
|
|
from cryptography.hazmat.primitives.asymmetric import ec
|
|
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
|
|
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
|
|
from django.conf import settings
|
|
from django.core.serializers.json import DjangoJSONEncoder
|
|
from django.utils import timezone
|
|
from django.utils.encoding import force_str
|
|
from py_vapid import Vapid, sign as vapid_sign
|
|
import requests
|
|
|
|
from .models import WebPushSubscription
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def b64_urlsafe_decode(data) -> bytes:
|
|
"""Décode une chaîne ou des octets base64url en complétant le padding si nécessaire."""
|
|
if isinstance(data, str):
|
|
data = data.encode('ascii')
|
|
rem = len(data) % 4
|
|
if rem > 0:
|
|
data += b'=' * (4 - rem)
|
|
return base64.urlsafe_b64decode(data)
|
|
|
|
|
|
def encrypt_webpush_rfc8291(plaintext: bytes, receiver_pub_bytes: bytes, auth_secret: bytes, salt: bytes = None) -> bytes:
|
|
"""
|
|
Chiffre une charge utile selon la norme Web Push RFC 8291 / RFC 8188 (aes128gcm).
|
|
Utilise directement la bibliothèque standard `cryptography` pour une fiabilité absolue.
|
|
"""
|
|
if salt is None:
|
|
salt = os.urandom(16)
|
|
|
|
# 1. Clé éphémère du serveur (P-256)
|
|
server_priv = ec.generate_private_key(ec.SECP256R1())
|
|
server_pub = server_priv.public_key()
|
|
server_pub_bytes = server_pub.public_bytes(
|
|
encoding=serialization.Encoding.X962,
|
|
format=serialization.PublicFormat.UncompressedPoint,
|
|
)
|
|
|
|
# 2. Clé publique du client (navigateur) et calcul du secret partagé ECDH
|
|
receiver_pub = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), receiver_pub_bytes)
|
|
ecdh_secret = server_priv.exchange(ec.ECDH(), receiver_pub)
|
|
|
|
# 3. Dérivation IKM (Input Keying Material)
|
|
auth_info = b'WebPush: info\x00' + receiver_pub_bytes + server_pub_bytes
|
|
ikm = HKDF(
|
|
algorithm=hashes.SHA256(),
|
|
length=32,
|
|
salt=auth_secret,
|
|
info=auth_info,
|
|
).derive(ecdh_secret)
|
|
|
|
# 4. Dérivation CEK (Content Encryption Key)
|
|
cek_info = b'Content-Encoding: aes128gcm\x00'
|
|
cek = HKDF(
|
|
algorithm=hashes.SHA256(),
|
|
length=16,
|
|
salt=salt,
|
|
info=cek_info,
|
|
).derive(ikm)
|
|
|
|
# 5. Dérivation Nonce
|
|
nonce_info = b'Content-Encoding: nonce\x00'
|
|
nonce = HKDF(
|
|
algorithm=hashes.SHA256(),
|
|
length=12,
|
|
salt=salt,
|
|
info=nonce_info,
|
|
).derive(ikm)
|
|
|
|
# 6. En-tête de message RFC 8188: salt (16 octets) + rs (4 octets, 4096) + idlen (1 octet) + keyid (65 octets)
|
|
rs = 4096
|
|
header = salt + struct.pack('>I', rs) + bytes([len(server_pub_bytes)]) + server_pub_bytes
|
|
|
|
# 7. Chiffrement AES-GCM (avec marqueur de fin 0x02 pour le dernier bloc)
|
|
record = plaintext + b'\x02'
|
|
aesgcm = AESGCM(cek)
|
|
ciphertext = aesgcm.encrypt(nonce, record, associated_data=None)
|
|
|
|
return header + ciphertext
|
|
|
|
|
|
def parse_device_info(user_agent: str) -> dict:
|
|
"""
|
|
Analyse succincte du User-Agent pour extraire le navigateur et le système d'exploitation.
|
|
"""
|
|
if not user_agent:
|
|
return {
|
|
'browser': 'Navigateur inconnu',
|
|
'os': 'Système inconnu',
|
|
'device_name': 'Appareil inconnu',
|
|
}
|
|
|
|
ua = user_agent
|
|
|
|
# 1. OS Detection
|
|
if 'Windows' in ua:
|
|
os_name = 'Windows'
|
|
elif 'Android' in ua:
|
|
os_name = 'Android'
|
|
elif 'iPhone' in ua or 'iPad' in ua or 'iPod' in ua:
|
|
os_name = 'iOS'
|
|
elif 'Macintosh' in ua or 'Mac OS' in ua:
|
|
os_name = 'macOS'
|
|
elif 'Linux' in ua:
|
|
os_name = 'Linux'
|
|
else:
|
|
os_name = 'Autre OS'
|
|
|
|
# 2. Browser Detection
|
|
if 'Edg/' in ua or 'Edge/' in ua:
|
|
browser_name = 'Microsoft Edge'
|
|
elif 'OPR/' in ua or 'Opera/' in ua:
|
|
browser_name = 'Opera'
|
|
elif 'Chrome/' in ua and 'Mobile' not in ua:
|
|
browser_name = 'Google Chrome'
|
|
elif 'Chrome/' in ua and 'Mobile' in ua:
|
|
browser_name = 'Chrome Mobile'
|
|
elif 'Safari/' in ua and 'Chrome' not in ua:
|
|
browser_name = 'Safari'
|
|
elif 'Firefox/' in ua:
|
|
browser_name = 'Mozilla Firefox'
|
|
else:
|
|
browser_name = 'Navigateur Web'
|
|
|
|
device_name = f"{browser_name} sur {os_name}"
|
|
return {
|
|
'browser': browser_name,
|
|
'os': os_name,
|
|
'device_name': device_name,
|
|
}
|
|
|
|
|
|
def get_vapid_key_obj():
|
|
"""Charge l'objet Vapid pour la signature des requêtes push."""
|
|
priv_key = getattr(settings, 'WEBPUSH_VAPID_PRIVATE_KEY', None)
|
|
if not priv_key:
|
|
return None
|
|
|
|
if isinstance(priv_key, Vapid):
|
|
return priv_key
|
|
|
|
try:
|
|
if isinstance(priv_key, str):
|
|
# Nettoyer les sauts de ligne échappés provenant des variables d'environnement
|
|
cleaned_key = priv_key.replace('\\n', '\n').strip()
|
|
if 'BEGIN PRIVATE KEY' in cleaned_key or 'BEGIN EC PRIVATE KEY' in cleaned_key:
|
|
return Vapid.from_pem(cleaned_key.encode('utf-8'))
|
|
elif os.path.exists(cleaned_key):
|
|
return Vapid.from_file(cleaned_key)
|
|
else:
|
|
return Vapid.from_string(cleaned_key)
|
|
elif isinstance(priv_key, bytes):
|
|
return Vapid.from_pem(priv_key)
|
|
except Exception as exc:
|
|
logger.error("Impossible de charger la clé privée VAPID: %s", exc)
|
|
return None
|
|
|
|
|
|
def get_vapid_claims():
|
|
"""Retourne les claims requis par la norme VAPID."""
|
|
admin_email = getattr(settings, 'WEBPUSH_VAPID_ADMIN_EMAIL', 'mailto:admin@valogeo.be')
|
|
if not admin_email.startswith('mailto:') and '@' in admin_email:
|
|
admin_email = f"mailto:{admin_email}"
|
|
return {'sub': admin_email}
|
|
|
|
|
|
def get_vapid_headers(endpoint: str, vapid_obj) -> dict:
|
|
"""
|
|
Génère les en-têtes d'authentification VAPID compatibles avec tous les fournisseurs Web Push
|
|
(Google FCM, Mozilla, Apple, Microsoft).
|
|
"""
|
|
parsed_url = urlparse(endpoint)
|
|
aud = f"{parsed_url.scheme}://{parsed_url.netloc}"
|
|
|
|
claims = get_vapid_claims()
|
|
claims['aud'] = aud
|
|
claims['exp'] = int(time.time()) + (12 * 3600)
|
|
|
|
pub_bytes = vapid_obj.public_key.public_bytes(
|
|
serialization.Encoding.X962,
|
|
serialization.PublicFormat.UncompressedPoint,
|
|
)
|
|
pub_b64 = base64.urlsafe_b64encode(pub_bytes).decode('ascii').rstrip('=')
|
|
|
|
jwt_token = vapid_sign(vapid_obj._base_sign(claims), vapid_obj.private_key)
|
|
|
|
return {
|
|
'Authorization': f'vapid t={jwt_token}, k={pub_b64}',
|
|
'Crypto-Key': f'p256ecdsa={pub_b64}',
|
|
}
|
|
|
|
|
|
def _send_single_push(subscription: WebPushSubscription, payload: dict, vapid_obj):
|
|
"""Envoie un message push chiffré à un abonnement individuel."""
|
|
try:
|
|
endpoint = subscription.endpoint
|
|
p256dh = subscription.p256dh
|
|
auth = subscription.auth
|
|
|
|
if not endpoint or not p256dh or not auth:
|
|
logger.warning("Abonnement WebPush incomplet pour %s (ID %s)", subscription.user.username, subscription.pk)
|
|
return False, "Abonnement incomplet"
|
|
|
|
receiver_pub_bytes = b64_urlsafe_decode(p256dh)
|
|
auth_secret = b64_urlsafe_decode(auth)
|
|
|
|
payload_bytes = json.dumps(payload, cls=DjangoJSONEncoder).encode('utf-8')
|
|
encrypted_body = encrypt_webpush_rfc8291(payload_bytes, receiver_pub_bytes, auth_secret)
|
|
|
|
vapid_headers = get_vapid_headers(endpoint, vapid_obj)
|
|
headers = {
|
|
**vapid_headers,
|
|
'Content-Type': 'application/octet-stream',
|
|
'Content-Encoding': 'aes128gcm',
|
|
'TTL': '86400',
|
|
'Urgency': 'high',
|
|
}
|
|
|
|
resp = requests.post(endpoint, data=encrypted_body, headers=headers, timeout=8.0)
|
|
|
|
if resp.status_code in (200, 201, 202):
|
|
logger.info("Notification push envoyée avec succès à %s (%s)", subscription.user.username, subscription.device_name)
|
|
return True, f"HTTP {resp.status_code}"
|
|
elif resp.status_code in (404, 410):
|
|
# L'abonnement a expiré ou a été révoqué par le navigateur
|
|
logger.info("Suppression de l'abonnement push expiré pour %s (HTTP %s)", subscription.user.username, resp.status_code)
|
|
try:
|
|
subscription.delete()
|
|
except Exception:
|
|
pass
|
|
return False, f"Abonnement expiré (HTTP {resp.status_code})"
|
|
else:
|
|
logger.warning("Erreur envoi WebPush pour %s (HTTP %s): %s", subscription.user.username, resp.status_code, resp.text[:200])
|
|
return False, f"HTTP {resp.status_code}: {resp.text[:100]}"
|
|
except Exception as exc:
|
|
logger.warning("Erreur inattendue envoi WebPush pour %s: %s", subscription.user.username, exc)
|
|
return False, str(exc)
|
|
|
|
|
|
def send_webpush_to_user(
|
|
user,
|
|
title: str,
|
|
body: str = '',
|
|
action_url: str = '',
|
|
tag: str = None,
|
|
icon: str = None,
|
|
badge: str = None,
|
|
extra_data: dict = None,
|
|
async_send: bool = True,
|
|
):
|
|
"""
|
|
Envoie une notification push à tous les appareils enregistrés pour un utilisateur.
|
|
Retourne un dictionnaire de résultat : {'success': int, 'failed': int, 'errors': list}.
|
|
"""
|
|
if not user or not user.is_authenticated:
|
|
return {'success': 0, 'failed': 0, 'errors': ['Utilisateur non authentifié']}
|
|
|
|
subscriptions = list(WebPushSubscription.objects.filter(user=user))
|
|
if not subscriptions:
|
|
return {'success': 0, 'failed': 0, 'errors': ['Aucun abonnement trouvé']}
|
|
|
|
vapid_obj = get_vapid_key_obj()
|
|
if not vapid_obj:
|
|
logger.warning("Envoi push ignoré : clé VAPID non configurée.")
|
|
return {'success': 0, 'failed': 0, 'errors': ['Clé VAPID non configurée']}
|
|
|
|
payload = {
|
|
'title': force_str(title),
|
|
'body': force_str(body or ''),
|
|
'icon': icon or '/static/common/android-192.png',
|
|
'badge': badge or '/static/common/favicon-48.png',
|
|
'action_url': force_str(action_url or '/notifications/'),
|
|
'tag': tag or f"loko-{int(timezone.now().timestamp())}",
|
|
'data': {
|
|
'url': force_str(action_url or '/notifications/'),
|
|
**(extra_data or {}),
|
|
},
|
|
}
|
|
|
|
results = {'success': 0, 'failed': 0, 'errors': []}
|
|
|
|
def _worker():
|
|
for sub in subscriptions:
|
|
ok, msg = _send_single_push(sub, payload, vapid_obj)
|
|
if ok:
|
|
results['success'] += 1
|
|
else:
|
|
results['failed'] += 1
|
|
results['errors'].append(f"{sub.device_name or 'Appareil'}: {msg}")
|
|
return results
|
|
|
|
if async_send:
|
|
t = Thread(target=_worker, daemon=True)
|
|
t.start()
|
|
return {'status': 'dispatched', 'count': len(subscriptions)}
|
|
else:
|
|
return _worker()
|
|
|
|
|
|
def send_webpush_to_users(
|
|
users,
|
|
title: str,
|
|
body: str = '',
|
|
action_url: str = '',
|
|
tag: str = None,
|
|
icon: str = None,
|
|
badge: str = None,
|
|
extra_data: dict = None,
|
|
async_send: bool = True,
|
|
):
|
|
"""
|
|
Envoie une notification push à une liste d'utilisateurs.
|
|
"""
|
|
if not users:
|
|
return {'success': 0, 'failed': 0, 'errors': []}
|
|
|
|
user_ids = [u.pk if hasattr(u, 'pk') else u for u in users if u]
|
|
subscriptions = list(WebPushSubscription.objects.filter(user_id__in=user_ids).select_related('user'))
|
|
if not subscriptions:
|
|
return {'success': 0, 'failed': 0, 'errors': []}
|
|
|
|
vapid_obj = get_vapid_key_obj()
|
|
if not vapid_obj:
|
|
logger.warning("Envoi push ignoré : clé VAPID non configurée.")
|
|
return {'success': 0, 'failed': 0, 'errors': ['Clé VAPID non configurée']}
|
|
|
|
payload = {
|
|
'title': force_str(title),
|
|
'body': force_str(body or ''),
|
|
'icon': icon or '/static/common/android-192.png',
|
|
'badge': badge or '/static/common/favicon-48.png',
|
|
'action_url': force_str(action_url or '/notifications/'),
|
|
'tag': tag or f"loko-{int(timezone.now().timestamp())}",
|
|
'data': {
|
|
'url': force_str(action_url or '/notifications/'),
|
|
**(extra_data or {}),
|
|
},
|
|
}
|
|
|
|
def _worker():
|
|
for sub in subscriptions:
|
|
_send_single_push(sub, payload, vapid_obj)
|
|
|
|
if async_send:
|
|
t = Thread(target=_worker, daemon=True)
|
|
t.start()
|
|
else:
|
|
_worker()
|