Coverage for app/backend/src/couchers/notifications/web_push_api.py: 78%

1import logging 

2from time import time 

3from typing import Any 

4from urllib.parse import urlparse 

5 

6import http_ece 

7import requests 

8from cryptography.hazmat.primitives import serialization 

9from cryptography.hazmat.primitives.asymmetric import ec 

10from py_vapid import Vapid 

11 

12from couchers.crypto import b64decode_unpadded, b64encode_unpadded 

13 

14logger = logging.getLogger(__name__) 

15 

16 

17def gen_vapid_keys() -> tuple[str, str]: 

18 prv_key = ec.generate_private_key(ec.SECP256R1()) 

19 pub_key = prv_key.public_key() 

20 prv = prv_key.private_numbers().private_value.to_bytes(length=32) 

21 pub = pub_key.public_bytes(serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint) 

22 return b64encode_unpadded(prv), b64encode_unpadded(pub) 

23 

24 

25def get_vapid_public_key_from_private_key(private: str) -> str: 

26 pub = Vapid.from_string(private).public_key 

27 result: str = b64encode_unpadded( 

28 pub.public_bytes(serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint) 

29 ) 

30 return result 

31 

32 

33def generate_vapid_authorization(endpoint: str, vapid_sub: str, vapid_private_key: str) -> str: 

34 url = urlparse(endpoint) 

35 vapid_claim = { 

36 "sub": vapid_sub, 

37 "aud": f"{url.scheme}://{url.netloc}", 

38 "exp": int(time()) + (12 * 60 * 60), 

39 } 

40 return Vapid.from_string(private_key=vapid_private_key).sign(vapid_claim)["Authorization"] # type: ignore[no-any-return] 

41 

42 

43def send_web_push( 

44 data: bytes, 

45 endpoint: str, 

46 auth_key: bytes, 

47 receiver_key: bytes, 

48 vapid_sub: str, 

49 vapid_private_key: str, 

50 ttl: int = 0, 

51) -> requests.Response: 

52 logger.debug(f"Sending {len(data)} bytes to {endpoint[:20]}...") 

53 headers = { 

54 "authorization": generate_vapid_authorization(endpoint, vapid_sub, vapid_private_key), 

55 "content-encoding": "aes128gcm", 

56 "ttl": str(ttl), 

57 # WNS (Windows Notification Service, the push backend for Edge) 400s with 

58 # "Ttl value conflicts with X-WNS-Cache-Policy" unless this matches the ttl 

59 "x-wns-cache-policy": "no-cache" if ttl == 0 else "cache", 

60 } 

61 

62 encrypted = http_ece.encrypt( 

63 data, 

64 private_key=ec.generate_private_key(ec.SECP256R1()), 

65 auth_secret=auth_key, 

66 dh=receiver_key, 

67 ) 

68 

69 return requests.post( 

70 endpoint, 

71 timeout=20, 

72 data=encrypted, 

73 headers=headers, 

74 ) 

75 

76 

77def debug_response_headers(resp: requests.Response) -> dict[str, str] | None: 

78 """Pick out the response headers worth reporting when a push fails. 

79 

80 WNS reports why it rejected a push in x-wns-* headers and leaves the body empty. 

81 """ 

82 headers = {k: v for k, v in resp.headers.items() if k.lower().startswith("x-wns-")} 

83 return headers or None 

84 

85 

86def decode_key(value: str) -> bytes: 

87 return b64decode_unpadded(value.encode()) 

88 

89 

90def parse_subscription_info(subscription_info: dict[str, Any]) -> tuple[str, bytes, bytes]: 

91 endpoint = subscription_info["endpoint"] 

92 auth_key = decode_key(subscription_info["keys"]["auth"]) 

93 receiver_key = decode_key(subscription_info["keys"]["p256dh"]) 

94 return endpoint, auth_key, receiver_key