odoo_source/addons/mail/web_push.py

# -*- coding: utf-8 -*-
# Part of Odoo. See LICENSE file for full copyright and licensing details.
import base64
import binascii
import json
import logging as logger
import os
import struct
import textwrap
import time

from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes, serialization
from cryptography.hazmat.primitives.asymmetric import ec, utils
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
from cryptography.hazmat.primitives.kdf.hkdf import HKDF
from cryptography.hazmat.primitives.serialization import Encoding, PublicFormat
from urllib.parse import urlparse

MAX_PAYLOAD_SIZE = 4096

_logger = logger.getLogger(__name__)

def _base64_decode_with_padding(value):
    return base64.urlsafe_b64decode(value + '==')

def generate_web_push_vapid_key():
    """
    Generate the VAPID (Voluntary Application Server Identification) used for the Web Push
    This function generates a signing key pair usable with the Elliptic Curve Digital
    Signature Algorithm (ECDSA) over the P-256 curve.
    These keys will be used during communication with the endpoint/browser
    https://www.rfc-editor.org/rfc/rfc8292
    """
    private_key = ec.generate_private_key(ec.SECP256R1(), default_backend())
    private_int = private_key.private_numbers().private_value
    private = private_int.to_bytes(32, 'big')
    private_string = base64.urlsafe_b64encode(private).decode('ascii').strip('=')

    public_key = private_key.public_key()
    public = public_key.public_bytes(
        encoding=serialization.Encoding.X962,
        format=serialization.PublicFormat.UncompressedPoint
    )
    public_string = base64.urlsafe_b64encode(public).decode('ascii').strip('=')
    return private_string, public_string

def _generate_jwt(endpoint, base_url, vapid_private_key):
    """
    JWT are a pair of JSON objects, turned into base64 strings, and signed with the private ECDH key
    https://www.rfc-editor.org/rfc/rfc7519
    https://www.rfc-editor.org/rfc/rfc8291
    :param endpoint: the browser endpoint
    :param base_url: the base url
    :param vapid_private_key: the private ECDH key generate at mail_entreprise install
    :return:
    """
    url = urlparse(endpoint)

    jwt_info = base64.urlsafe_b64encode(json.dumps({
        'typ': 'JWT',
        'alg': 'ES256'
    }).encode())

    # The expiration is a timestamp in seconds and must be no longer 12 hours.
    token_validity = 12 * 60 * 60

    jwt_data = base64.urlsafe_b64encode(json.dumps({
        # aud: The “Audience” is a JWT construct that indicates the recipient scheme and host
        # e.g. for an endpoint like https://updates.push.services.mozilla.com/wpush/v2/gAAAAABY...,
        #      the “aud” would be https://updates.push.services.mozilla.com
        'aud': '{}://{}'.format(url.scheme, url.netloc),
        # sub: the sub value needs to be either a URL address. This is so that if a push service needed to reach out
        # to sender, it can find contact information from the JWT.
        'sub': base_url,
        # exp: It's the expiration of the JWT, this prevents snoopers from being able to re-use a JWT if they intercept it.
        'exp': int(time.time()) + token_validity
    }).encode())

    unsigned_token = '{}.{}'.format(jwt_info.decode().strip('='), jwt_data.decode().strip('='))

    # Retrieve the private key using a P256 elliptic curve
    vapid_private_key_decoded = _base64_decode_with_padding(vapid_private_key)
    private_key = ec.derive_private_key(int(binascii.hexlify(vapid_private_key_decoded), 16), ec.SECP256R1(), default_backend())

    # sign with ECDSA SHA-256
    signature = private_key.sign(unsigned_token.encode(), ec.ECDSA(hashes.SHA256()))
    (r, s) = utils.decode_dss_signature(signature)
    sig = base64.urlsafe_b64encode(r.to_bytes(32, 'big') + s.to_bytes(32, 'big'))

    return '{}.{}'.format(unsigned_token, sig.decode().strip('='))

def _iv(base, counter):
    mask = int.from_bytes(base[4:], 'big')
    return base[:4] + (counter ^ mask).to_bytes(8, 'big')

def _derive_key(salt, private_key, device):
    # browser keys
    device_keys = json.loads(device["keys"])
    p256dh = _base64_decode_with_padding(device_keys.get('p256dh'))
    auth = _base64_decode_with_padding(device_keys.get('auth'))

    # generate a public key derived from the browser public key
    pub_key = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), p256dh)
    sender_pub_key = private_key.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint
    )

    context = b"WebPush: info\x00" + p256dh + sender_pub_key
    key_info = b"Content-Encoding: aes128gcm\x00"
    nonce_info = b"Content-Encoding: nonce\x00"

    # Create the 3 HKDF keys needed to encrypt the message (auth, key, nonce)
    hkdf_auth = HKDF(
        algorithm=hashes.SHA256(),
        length=32,
        salt=auth,
        info=context,
        backend=default_backend(),
    )
    hkdf_key = HKDF(
        algorithm=hashes.SHA256(),
        length=16,
        salt=salt,
        info=key_info,
        backend=default_backend(),
    )
    hkdf_nonce = HKDF(
        algorithm=hashes.SHA256(),
        length=12,
        salt=salt,
        info=nonce_info,
        backend=default_backend(),
    )
    secret = hkdf_auth.derive(private_key.exchange(ec.ECDH(), pub_key))
    return hkdf_key.derive(secret), hkdf_nonce.derive(secret)

def _encrypt_payload(content, device, record_size=MAX_PAYLOAD_SIZE):
    """
    Encrypt a payload for Push Notification Endpoint using AES128GCM

    https://www.rfc-editor.org/rfc/rfc7516
    https://www.rfc-editor.org/rfc/rfc8188
    :param content: the unencrypted payload
    :param device: the web push user browser information
    :param record_size: record size must be bigger than 18
    :return: the encrypted payload
    """
    # The private_key is an ephemeral ECDH key used only for a transaction
    private_key = ec.generate_private_key(ec.SECP256R1(), default_backend())
    salt = os.urandom(16)
    # generate key
    (key, nonce) = _derive_key(salt=salt, private_key=private_key, device=device)
    # AEAD_AES_128_GCM produces ciphertext 16 octets longer than its input plaintext.
    # Therefore, the unencrypted content of each record is shorter than the record size by 16 octets.
    # Valid records always contain at least a padding delimiter octet and a 16-octet authentication tag.
    overhead = 1 + 16
    chunk_size = record_size - overhead

    body = b""
    end = len(content)
    aesgcm = AESGCM(key)
    for i in range(0, end, chunk_size):
        padding = b"\x02" if (i + chunk_size) >= end else b"\x01"
        body += aesgcm.encrypt(nonce, content[i: i + chunk_size] + padding, None)

    sender_public_key = private_key.public_key().public_bytes(
        Encoding.X962, PublicFormat.UncompressedPoint
    )

    # +-----------+-----------------+---------------------------+-------------------------------------------+
    # | salt (16) | record_size (4) | sender_public_key.len (1) | sender_public_key (sender_public_key.len) |
    # +-----------+-----------------+---------------------------+-------------------------------------------+
    header = struct.pack("!16sLB", salt, record_size, len(sender_public_key))
    header += sender_public_key
    return header + body

def push_to_end_point(base_url, device, payload, vapid_private_key, vapid_public_key, session):
    endpoint = device["endpoint"]
    jwt = _generate_jwt(endpoint, base_url, vapid_private_key)
    body_payload = payload.encode()
    payload = _encrypt_payload(body_payload, device)
    headers = {
        #  Authorization header field contains these parameters:
        #  - "t" is the JWT;
        #  - "k" the base64url-encoded key that signed that token.
        'Authorization': 'vapid t={}, k={}'.format(jwt, vapid_public_key),
        'Content-Encoding': 'aes128gcm',
        'TTL': '0',
    }

    response = session.post(endpoint, headers=headers, data=payload, timeout=5)
    if response.status_code == 201:
        _logger.debug('Sent push notification %s', endpoint)
    else:
        error_message_shorten = textwrap.shorten(response.text, 100)
        _logger.warning('Failed push notification %s %d - %s',
                        endpoint, response.status_code, error_message_shorten)

        # Invalid subscription
        if response.status_code == 404 or response.status_code == 410:
            raise DeviceUnreachableError("Device Unreachable")


class DeviceUnreachableError(Exception):
    pass