Coverage for src/lexigram/notification/backends/push/web_push.py: 29%

1"""Web Push (RFC 8030) notification backend using VAPID + encryption.""" 

2 

3from __future__ import annotations 

4 

5import asyncio 

6from typing import TYPE_CHECKING 

7import uuid 

8 

9from lexigram.contracts.core import HealthCheckResult, HealthStatus 

10from lexigram.contracts.mailer import MessageDeliveryReceipt 

11from lexigram.contracts.notification.types import PushMessage 

12from lexigram.logging import get_logger 

13from lexigram.notification.exceptions import WebPushNotificationError 

14from lexigram.result import Err, Ok, Result 

15 

16if TYPE_CHECKING: 

17 from lexigram.contracts.notification.errors import NotificationError 

18 

19logger = get_logger(__name__) 

20 

21try: 

22 from pywebpush import WebPushException, webpush # type: ignore[import-untyped] 

23except ImportError: 

24 WebPushException = Exception 

25 

26 def webpush(*args: object, **kwargs: object) -> None: 

27 raise ImportError( 

28 "pywebpush is not installed. Install lexigram-notification[web-push]." 

29 ) 

30 

31 

32_HTTP_CLIENT_TIMEOUT = 30 

33 

34 

35class WebPushChannel: 

36 """Web Push (RFC 8030) notification backend. 

37 

38 Implements :class:`~lexigram.contracts.notification.protocols.PushChannelProtocol`. 

39 Uses VAPID (RFC 8292) for application-level authentication and RFC 8291 

40 (aes128gcm) for message encryption. 

41 

42 Requires the ``pywebpush`` optional extra. 

43 

44 Args: 

45 vapid_private_key: VAPID private key (PEM-encoded EC prime256v1). 

46 vapid_public_key: VAPID public key (base64url-encoded). 

47 vapid_claims_subject: Contact URI for VAPID claims 

48 (e.g. ``"mailto:ops@example.com"``). 

49 http_timeout: HTTP request timeout in seconds. 

50 """ 

51 

52 def __init__( 

53 self, 

54 vapid_private_key: str, 

55 vapid_public_key: str, 

56 vapid_claims_subject: str, 

57 http_timeout: int = _HTTP_CLIENT_TIMEOUT, 

58 ) -> None: 

59 self._vapid_private_key = vapid_private_key 

60 self._vapid_public_key = vapid_public_key 

61 self._vapid_claims_subject = vapid_claims_subject 

62 self._http_timeout = http_timeout 

63 

64 async def send( 

65 self, message: PushMessage 

66 ) -> Result[MessageDeliveryReceipt, NotificationError]: 

67 """Send a Web Push notification. 

68 

69 ``message.to`` contains a list of **endpoint URLs** (not device tokens). 

70 Each endpoint is a separate push subscription. 

71 

72 Args: 

73 message: The push notification to deliver. ``message.to`` must 

74 contain at least one endpoint URL. 

75 

76 Returns: 

77 Ok(MessageDeliveryReceipt) on success. 

78 Err(WebPushNotificationError) for delivery failures. 

79 """ 

80 if not message.to: 

81 raise ValueError("WebPushChannel.send() requires at least one endpoint") 

82 

83 from lexigram.serialization import dumps_str 

84 

85 payload: dict[str, object] = { 

86 "title": message.title, 

87 "body": message.body, 

88 } 

89 if message.data: 

90 payload["data"] = message.data 

91 if message.badge is not None: 

92 payload["badge"] = message.badge 

93 if message.image: 

94 payload["icon"] = message.image 

95 

96 endpoint = message.to[0] 

97 

98 try: 

99 subscription_info: dict[str, object] = {"endpoint": endpoint} 

100 

101 keys = (message.data or {}).get("keys") 

102 if isinstance(keys, dict): 

103 p256dh = keys.get("p256dh", "") 

104 auth = keys.get("auth", "") 

105 if p256dh and auth: 

106 subscription_info["keys"] = {"p256dh": p256dh, "auth": auth} 

107 

108 result = webpush( 

109 subscription_info=subscription_info, 

110 data=dumps_str(payload), 

111 vapid_private_key=self._vapid_private_key, 

112 vapid_public_key=self._vapid_public_key, 

113 vapid_claims={"sub": self._vapid_claims_subject}, 

114 ttl=message.ttl or 0, 

115 timeout=self._http_timeout, 

116 ) 

117 

118 receipt = MessageDeliveryReceipt( 

119 message_id=str(uuid.uuid4()), 

120 backend="web_push", 

121 channel="push", 

122 provider_reference=endpoint[:64], 

123 ) 

124 logger.info( 

125 "webpush_sent", 

126 endpoint_prefix=endpoint[:48], 

127 status_code=result.status_code 

128 if hasattr(result, "status_code") 

129 else None, 

130 ) 

131 return Ok(receipt) 

132 

133 except WebPushException as exc: 

134 status_code = getattr(exc, "status_code", None) 

135 response_body = getattr(exc, "response_body", None) 

136 logger.warning( 

137 "webpush_send_failed", 

138 endpoint_prefix=endpoint[:48], 

139 status_code=status_code, 

140 response=response_body, 

141 ) 

142 

143 if status_code in (410, 404): 

144 return Err( 

145 WebPushNotificationError( 

146 message=f"Subscription gone (HTTP {status_code})", 

147 status_code=status_code, 

148 reason="subscription_gone", 

149 ) 

150 ) 

151 

152 return Err( 

153 WebPushNotificationError( 

154 message=str(exc), 

155 status_code=status_code or 0, 

156 reason=response_body or "webpush_error", 

157 ) 

158 ) 

159 

160 async def send_batch( 

161 self, messages: list[PushMessage] 

162 ) -> list[Result[MessageDeliveryReceipt, NotificationError]]: 

163 """Send multiple Web Push notifications concurrently. 

164 

165 Args: 

166 messages: Notifications to deliver. 

167 

168 Returns: 

169 List of Result values, one per message, order preserved. 

170 """ 

171 tasks = [self.send(msg) for msg in messages] 

172 return await asyncio.gather(*tasks) 

173 

174 async def health_check(self, timeout: float = 5.0) -> HealthCheckResult: 

175 """Check Web Push infrastructure availability. 

176 

177 Since Web Push has no global health endpoint, this is a best-effort 

178 check that verifies the VAPID key is loadable. 

179 

180 Args: 

181 timeout: Max seconds to wait (unused, but kept for protocol compat). 

182 

183 Returns: 

184 HealthCheckResult. 

185 """ 

186 try: 

187 from cryptography.hazmat.primitives.asymmetric import ec 

188 from py_vapid import Vapid # type: ignore[import-untyped] 

189 

190 v = Vapid.from_string(self._vapid_private_key) 

191 private_key_obj = v.private_key 

192 if not isinstance(private_key_obj, ec.EllipticCurvePrivateKey): 

193 return HealthCheckResult( 

194 component="web_push", 

195 status=HealthStatus.UNHEALTHY, 

196 details={"error": "VAPID key is not an EC key"}, 

197 ) 

198 return HealthCheckResult( 

199 component="web_push", 

200 status=HealthStatus.HEALTHY, 

201 details={"vapid_key_valid": True}, 

202 ) 

203 except Exception as exc: # noqa: BLE001 

204 return HealthCheckResult( 

205 component="web_push", 

206 status=HealthStatus.UNHEALTHY, 

207 details={"error": str(exc)}, 

208 ) 

209 

210 

211__all__ = ["WebPushChannel"]