1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
|
"""
Web Push to a phone: RFC 8291 encryption and the one outbound request.
The Android application registers with a UnifiedPush distributor (ntfy, or any
other) and hands the hub an endpoint URL and a P-256 key; the hub POSTs each
notification there, encrypted to that key (`docs/MESHBAY_DESIGN.md` §7.6). The
push server relays bytes it cannot read; what it does learn is *when* this
person is notified, which is the same metadata the hub already holds.
**The endpoint is a URL a member supplied, and the hub fetches it.** That is
the shape of an SSRF, so a send resolves the host itself, refuses unless every
address is public, and connects to the address it checked — the hostname rides
only as the TLS server name and the Host header, so a second resolution cannot
point the request somewhere else. No redirect is followed.
"""
import asyncio
import base64
import ipaddress
import json
import logging
import os
import socket
from dataclasses import dataclass
from urllib.parse import urlsplit, urlunsplit
import httpx
from cryptography.hazmat.primitives import hashes, hmac, serialization
from cryptography.hazmat.primitives.asymmetric import ec
from cryptography.hazmat.primitives.ciphers.aead import AESGCM
log = logging.getLogger(__name__)
RECORD_SIZE = 4096
SEND_TIMEOUT = 5.0
MAX_ENDPOINT = 1024
# RFC 8030 §5.2: a push service may keep a message this long for a phone that is
# off. A chat line an hour old is still worth seeing; one a day old is not.
TTL_CHAT = 3600
TTL_OTHER = 86400
class EndpointRefused(ValueError):
"""The endpoint is not one the hub will send to."""
def b64url_decode(value: str) -> bytes:
return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
def _hmac(key: bytes, data: bytes) -> bytes:
h = hmac.HMAC(key, hashes.SHA256())
h.update(data)
return h.finalize()
def check_keys(p256dh: str, auth: str) -> tuple[bytes, bytes]:
"""Decode and validate a subscription's keys; ValueError when they are not."""
try:
ua_public = b64url_decode(p256dh)
auth_secret = b64url_decode(auth)
except (ValueError, TypeError) as e:
raise ValueError("keys are not base64url") from e
if len(ua_public) != 65 or ua_public[0] != 4:
raise ValueError("p256dh is not an uncompressed P-256 point")
if len(auth_secret) != 16:
raise ValueError("auth is not 16 bytes")
# Raises ValueError for a point that is not on the curve.
ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), ua_public)
return ua_public, auth_secret
def encrypt(plaintext: bytes, ua_public: bytes, auth_secret: bytes, *,
as_private: ec.EllipticCurvePrivateKey | None = None,
salt: bytes | None = None) -> bytes:
"""One aes128gcm record (RFC 8188) keyed as RFC 8291 §3.4 says.
`as_private` and `salt` are parameters only so the RFC's own example can be
replayed; a send always draws both fresh.
"""
if len(plaintext) > RECORD_SIZE - 16 - 1 - 86:
raise ValueError("push payload too large for one record")
as_private = as_private or ec.generate_private_key(ec.SECP256R1())
salt = salt or os.urandom(16)
as_public = as_private.public_key().public_bytes(
serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint)
ua_key = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), ua_public)
ecdh_secret = as_private.exchange(ec.ECDH(), ua_key)
prk_key = _hmac(auth_secret, ecdh_secret)
key_info = b"WebPush: info\x00" + ua_public + as_public
ikm = _hmac(prk_key, key_info + b"\x01")
prk = _hmac(salt, ikm)
cek = _hmac(prk, b"Content-Encoding: aes128gcm\x00\x01")[:16]
nonce = _hmac(prk, b"Content-Encoding: nonce\x00\x01")[:12]
header = salt + RECORD_SIZE.to_bytes(4, "big") + bytes([len(as_public)]) + as_public
return header + AESGCM(cek).encrypt(nonce, plaintext + b"\x02", None)
def check_endpoint(url: str) -> tuple[str, int]:
"""The endpoint's shape, checked when it is registered: https, a host, no
credentials, not an address that is private on its face. Where its name
resolves is checked at every send, because that can change."""
if len(url) > MAX_ENDPOINT:
raise EndpointRefused("endpoint too long")
parts = urlsplit(url)
if parts.scheme != "https" or not parts.hostname:
raise EndpointRefused("endpoint must be an https URL")
if parts.username or parts.password:
raise EndpointRefused("endpoint must not carry credentials")
try:
port = parts.port or 443
except ValueError as e:
raise EndpointRefused("endpoint port is not a number") from e
host = parts.hostname
try:
literal = ipaddress.ip_address(host)
except ValueError:
literal = None
if literal is not None and not literal.is_global:
raise EndpointRefused("endpoint is not a public address")
return host, port
async def _resolve_public(host: str, port: int) -> str:
infos = await asyncio.get_running_loop().getaddrinfo(
host, port, type=socket.SOCK_STREAM)
addresses = {info[4][0] for info in infos}
if not addresses:
raise EndpointRefused("endpoint does not resolve")
for a in addresses:
if not ipaddress.ip_address(a.split("%", 1)[0]).is_global:
raise EndpointRefused("endpoint resolves to a non-public address")
# IPv4 first: a hub with an AAAA answer and no IPv6 route is common.
return sorted(addresses, key=lambda a: (":" in a, a))[0]
@dataclass
class Target:
endpoint: str
p256dh: str
auth: str
# Outcomes of one send, for the caller to act on.
DELIVERED = "delivered"
GONE = "gone" # 404/410: the registration no longer exists (RFC 8030 §7.3)
FAILED = "failed"
async def send(target: Target, payload: dict, *, ttl: int,
client: httpx.AsyncClient | None = None) -> str:
"""Encrypt `payload` for one subscription and POST it. Never raises."""
try:
host, port = check_endpoint(target.endpoint)
ua_public, auth_secret = check_keys(target.p256dh, target.auth)
body = encrypt(json.dumps(payload, separators=(",", ":")).encode(),
ua_public, auth_secret)
address = await _resolve_public(host, port)
except (EndpointRefused, ValueError, OSError) as e:
log.info("push refused before sending: %s", e)
return FAILED
parts = urlsplit(target.endpoint)
netloc = f"[{address}]" if ":" in address else address
if parts.port:
netloc += f":{parts.port}"
pinned = urlunsplit((parts.scheme, netloc, parts.path or "/", parts.query, ""))
headers = {
"Host": parts.netloc,
"Content-Encoding": "aes128gcm",
"Content-Type": "application/octet-stream",
"TTL": str(ttl),
"Urgency": "normal",
}
own = client is None
client = client or httpx.AsyncClient(timeout=SEND_TIMEOUT, follow_redirects=False)
try:
resp = await client.post(pinned, content=body, headers=headers,
extensions={"sni_hostname": host})
except httpx.HTTPError as e:
# The URL path is a bearer capability for this phone: never logged.
log.info("push to %s failed: %s", host, type(e).__name__)
return FAILED
finally:
if own:
await client.aclose()
if resp.status_code in (404, 410):
return GONE
if 200 <= resp.status_code < 300:
return DELIVERED
log.info("push to %s answered %d", host, resp.status_code)
return FAILED
|