#!/usr/bin/env python3 """Web Push for the INBOX (inbox.html on a phone's Home Screen / desktop Chrome): the sender side, used by inbox_server.py (deployed next to it as /srv/stickbash-inbox/inbox_push.py). Needs the `cryptography` package (the VM's python3-cryptography); inbox_server.py itself stays stdlib-only and switches push off when this cannot be imported. RFC 8291 message encryption (aes128gcm, RFC 8188) + RFC 8292 VAPID (ES256 JWT), built from the `cryptography` package's primitives (ECDH P-256, HKDF/HMAC-SHA256, AES-128-GCM, ECDSA) — no hand-written ciphers. self_test() checks the encryption against the RFC 8291 appendix A example byte for byte. Files in the inbox dir (INBOX_DIR or this script's dir), both 0600, never printed, never committed: vapid_private.pem the VAPID signing key (made by `inbox_server.py push-init` on the VM) push.json {subs:[{endpoint, keys:{p256dh, auth}, ts, ua, ok, fail}], queue:[events], sent:{lane: ts}} python3 inbox_push.py selftest RFC 8291 example + a VAPID JWT round trip (no network, no key files) """ import base64, hashlib, hmac, json, os, struct, time from urllib.parse import urlparse from urllib.request import Request, urlopen from urllib.error import HTTPError, URLError from cryptography.hazmat.primitives import hashes, serialization from cryptography.hazmat.primitives.asymmetric import ec from cryptography.hazmat.primitives.asymmetric.utils import decode_dss_signature, encode_dss_signature from cryptography.hazmat.primitives.ciphers.aead import AESGCM VAPID_SUB = 'https://bashatest.howincloud.com/inbox.html' # the VAPID contact claim (a URL: no e-mail address in pushes) RS = 4096 def b64u(b): return base64.urlsafe_b64encode(b).rstrip(b'=').decode() def unb64u(s): s = s.strip() return base64.urlsafe_b64decode(s + '=' * (-len(s) % 4)) def _hmac(key, msg): return hmac.new(key, msg, hashlib.sha256).digest() def _pub_bytes(pub): return pub.public_bytes(serialization.Encoding.X962, serialization.PublicFormat.UncompressedPoint) # ---------- RFC 8291: the message encryption ---------- def encrypt(plaintext, ua_public_b64, auth_b64, _as_private=None, _salt=None): """The aes128gcm body for one push message to the subscription with keys (p256dh, auth).""" ua_public = unb64u(ua_public_b64) auth = unb64u(auth_b64) if len(ua_public) != 65 or ua_public[0] != 4 or len(auth) < 16: raise ValueError('bad subscription keys') ua_key = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), ua_public) as_private = _as_private or ec.generate_private_key(ec.SECP256R1()) as_public = _pub_bytes(as_private.public_key()) ecdh = as_private.exchange(ec.ECDH(), ua_key) prk_key = _hmac(auth, ecdh) ikm = _hmac(prk_key, b'WebPush: info\x00' + ua_public + as_public + b'\x01') salt = _salt or os.urandom(16) prk = _hmac(salt, ikm) cek = _hmac(prk, b'Content-Encoding: aes128gcm\x00\x01')[:16] nonce = _hmac(prk, b'Content-Encoding: nonce\x00\x01')[:12] if len(plaintext) + 1 + 16 > RS: raise ValueError('payload too big') ct = AESGCM(cek).encrypt(nonce, plaintext + b'\x02', None) # one record: the 0x02 last-record delimiter, no padding return salt + struct.pack('!IB', RS, len(as_public)) + as_public + ct # ---------- RFC 8292: VAPID ---------- def new_vapid_pem(): k = ec.generate_private_key(ec.SECP256R1()) return k.private_bytes(serialization.Encoding.PEM, serialization.PrivateFormat.PKCS8, serialization.NoEncryption()) def load_vapid(path): with open(path, 'rb') as f: return serialization.load_pem_private_key(f.read(), password=None) def vapid_public_b64(priv): return b64u(_pub_bytes(priv.public_key())) def vapid_auth(priv, endpoint, now=None): """The Authorization header value for a POST to endpoint (aud = the push service's origin, 12 h expiry).""" u = urlparse(endpoint) claims = {'aud': '%s://%s' % (u.scheme, u.netloc), 'exp': int((now or time.time()) + 12 * 3600), 'sub': VAPID_SUB} head = b64u(json.dumps({'typ': 'JWT', 'alg': 'ES256'}, separators=(',', ':')).encode()) body = b64u(json.dumps(claims, separators=(',', ':')).encode()) signing = ('%s.%s' % (head, body)).encode() r, s = decode_dss_signature(priv.sign(signing, ec.ECDSA(hashes.SHA256()))) jwt = '%s.%s.%s' % (head, body, b64u(r.to_bytes(32, 'big') + s.to_bytes(32, 'big'))) return 'vapid t=%s, k=%s' % (jwt, vapid_public_b64(priv)) # ---------- sending ---------- # push services the page's subscriptions may point at (the server only ever POSTs to these hosts) PUSH_HOSTS = ('fcm.googleapis.com', 'android.googleapis.com', 'web.push.apple.com', 'push.services.mozilla.com', 'updates.push.services.mozilla.com', 'notify.windows.com') def endpoint_ok(endpoint): try: u = urlparse(endpoint) except ValueError: return False host = (u.hostname or '').lower() return u.scheme == 'https' and len(endpoint) <= 1024 and any(host == h or host.endswith('.' + h) for h in PUSH_HOSTS) def send(priv, sub, payload, ttl=86400, urgency='normal', timeout=15, _post=None): """POST one encrypted message. Returns the HTTP status (201/200 = accepted; 404/410 = the subscription is gone; 0 = network error). _post(url, body, headers) -> status replaces the network in tests.""" endpoint = sub['endpoint'] if not endpoint_ok(endpoint): return 400 data = json.dumps(payload, ensure_ascii=False, separators=(',', ':')).encode() body = encrypt(data, sub['keys']['p256dh'], sub['keys']['auth']) headers = {'Authorization': vapid_auth(priv, endpoint), 'Content-Encoding': 'aes128gcm', 'Content-Type': 'application/octet-stream', 'TTL': str(int(ttl)), 'Urgency': urgency} if _post: return _post(endpoint, body, headers) try: with urlopen(Request(endpoint, data=body, headers=headers, method='POST'), timeout=timeout) as r: return r.status except HTTPError as e: return e.code except (URLError, OSError): return 0 # ---------- self test ---------- def self_test(): # RFC 8291 appendix A plain = b'When I grow up, I want to be a watermelon' as_priv_d = int.from_bytes(unb64u('yfWPiYE-n46HLnH0KqZOF1fJJU3MYrct3AELtAQ-oRw'), 'big') as_priv = ec.derive_private_key(as_priv_d, ec.SECP256R1()) assert b64u(_pub_bytes(as_priv.public_key())) == \ 'BP4z9KsN6nGRTbVYI_c7VJSPQTBtkgcy27mlmlMoZIIgDll6e3vCYLocInmYWAmS6TlzAC8wEqKK6PBru3jl7A8' ua_pub = 'BCVxsr7N_eNgVRqvHtD0zTZsEc6-VV-JvLexhqUzORcxaOzi6-AYWXvTBHm4bjyPjs7Vd8pZGH6SRpkNtoIAiw4' out = encrypt(plain, ua_pub, 'BTBZMqHH6r4Tts7J_aSIgg', _as_private=as_priv, _salt=unb64u('DGv6ra1nlYgDCS1FRnbzlw')) want = ('DGv6ra1nlYgDCS1FRnbzlwAAEABBBP4z9KsN6nGRTbVYI_c7VJSPQTBtkgcy27mlmlMoZIIgDll6e3vCYLocInmYWAmS6TlzAC8wEqKK6PBru' '3jl7A_yl95bQpu6cVPTpK4Mqgkf1CXztLVBSt2Ks3oZwbuwXPXLWyouBWLVWGNWQexSgSxsj_Qulcy4a-fN') assert b64u(out) == want, 'RFC 8291 example mismatch' # VAPID: the JWT verifies with the public key it carries priv = serialization.load_pem_private_key(new_vapid_pem(), password=None) hv = vapid_auth(priv, 'https://fcm.googleapis.com/fcm/send/abc') t = hv.split('t=')[1].split(',')[0] k = hv.split('k=')[1] h, b, sig = t.split('.') raw = unb64u(sig) pub = ec.EllipticCurvePublicKey.from_encoded_point(ec.SECP256R1(), unb64u(k)) pub.verify(encode_dss_signature(int.from_bytes(raw[:32], 'big'), int.from_bytes(raw[32:], 'big')), ('%s.%s' % (h, b)).encode(), ec.ECDSA(hashes.SHA256())) claims = json.loads(unb64u(b)) assert claims['aud'] == 'https://fcm.googleapis.com' and claims['sub'] == VAPID_SUB and claims['exp'] > time.time() assert endpoint_ok('https://web.push.apple.com/QAbc') and endpoint_ok('https://fcm.googleapis.com/fcm/send/x') assert not endpoint_ok('http://fcm.googleapis.com/x') and not endpoint_ok('https://evil.example/fcm.googleapis.com') assert not endpoint_ok('https://fcm.googleapis.com.evil.example/x') return True if __name__ == '__main__': import sys if sys.argv[1:] == ['selftest']: self_test() print('inbox_push selftest: ok (RFC 8291 example matches, VAPID JWT verifies)') else: sys.exit(__doc__)