entrouvert
/
debian-pywebpush
17
0
Fork 0
Code Issues Pull Requests 1 Releases Activity

Merge pull request #2 from jrconlin/feature/tests 

feat: Added tests, restructured code
This commit is contained in:
JR Conlin 2016-04-27 09:54:48 -07:00
parent 5c134746ac 68ab34a3eb
commit 9dd9e7607d
9 changed files with 325 additions and 131 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
CHANGELOG.md
View File

@ -1,3 +1,7 @@
## 0.2 (2016-04-27)
feat: Added tests, restructured code
## 0.1 (2016-04-25)
Initial release

38
README.md
View File

@ -2,7 +2,7 @@
This is a work in progress.
## App Installation
## Installation
You'll need to run `python virtualenv`.
Then
@ -11,7 +11,39 @@ bin/pip install -r requirements.txt
bin/python setup.py develop
```
### App Usage
## Usage
`webpush/publish.py` contains an example in the stand-alone function.
In the browser, the promise handler for
[registration.pushManager.subscribe()](https://developer.mozilla.org/en-US/docs/Web/API/PushManager/subscribe)
returns a
[PushSubscription](https://developer.mozilla.org/en-US/docs/Web/API/PushSubscription)
object. This object has a .toJSON() method that will return a JSON
object that contains all the info we need to encrypt and push data.
As illustration, a subscription info object may look like:
```
{"endpoint": "https://updates.push.services.mozilla.com/push/v1/gAA...", "keys": {"auth": "k8J...", "p256dh": "BOr..."}}
```
How you send the PushSubscription data to your backend, store it
referenced to the user who requested it, and recall it when there's
new a new push subscription update is left as an excerise for the
reader.
The data can be any serial content (string, bit array, serialized
JSON, etc), but be sure that your receiving application is able to
parse and understand it. (e.g. `data = "Mary had a little lamb."`)
`headers` is a `dict`ionary of additional HTTP header values (e.g.
[VAPID](https://github.com/mozilla-services/vapid/tree/master/python)
self identification headers). It is optional and may be omitted.
to send:
```
WebPusher(subscription_info).send(data, headers)
```
You can also simply encode the data to send later by calling
```
encoded = WebPush(subscription_info).encode(data)
```

1
requirements.txt
View File

@ -1,2 +1,3 @@
http-ece==0.5.0
python-jose==0.5.6
requests==2.9.1

8
setup.cfg Normal file
View File

@ -0,0 +1,8 @@
[nosetests]
verbose=True
verbosity=1
cover-tests=True
cover-erase=True
with-coverage=True
detailed-errors=True
cover-package=webpush

32
setup.py
View File

@ -3,20 +3,30 @@ import os
from setuptools import setup
__version__ = "0.2"
def read_from(file):
reply = []
with io.open(os.path.join(here, file), encoding='utf8') as f:
for l in f:
l = l.strip()
if not l:
break
if l[0] != '#' or l[:2] != '//':
reply.append(l)
return reply
here = os.path.abspath(os.path.dirname(__file__))
with io.open(os.path.join(here, 'README.md'), encoding='utf8') as f:
README = f.read()
with io.open(os.path.join(here, 'CHANGELOG.md'), encoding='utf8') as f:
CHANGES = f.read()
extra_options = {
"packages": ["http-ece==0.5.0"]
}
setup(name="Webpusher",
version="0.1",
description='Webpush publication library',
setup(name="pywebpush",
version="0.2",
description='WebPush publication library',
long_description=README + '\n\n' + CHANGES,
classifiers=["Topic :: Internet :: WWW/HTTP",
"Programming Language :: Python :: Implementation :: PyPy",
@ -27,11 +37,11 @@ setup(name="Webpusher",
keywords='push webpush publication',
author="jr conlin",
author_email="src+webpusher@jrconlin.com",
url='http:///',
url='https://github.com/jrconlin/pywebpush',
license="MPL2",
test_suite="nose.collector",
include_package_data=True,
zip_safe=False,
tests_require=['nose', 'coverage', 'mock>=1.0.1', 'moto>=0.4.1'],
**extra_options
install_requires=read_from('requirements.txt'),
tests_require=read_from('test-requirements.txt')
)

3
test-requirements.txt Normal file
View File

@ -0,0 +1,3 @@
nose
coverage
mock>=1.0.1

145
webpush/__init__.py
View File

@ -0,0 +1,145 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import base64
import os
import http_ece
import pyelliptic
import requests
class WebPushException(Exception):
pass
class WebPusher:
"""WebPusher encrypts a data block using HTTP Encrypted Content Encoding
for WebPush.
See https://tools.ietf.org/html/draft-ietf-webpush-protocol-04
for the current specification, and
https://developer.mozilla.org/en-US/docs/Web/API/Push_API for an
overview of Web Push.
Example of use:
The javascript promise handler for PushManager.subscribe()
receives a subscription_info object. subscription_info.getJSON()
will return a JSON representation.
(e.g.
.. code-block:: javascript
subscription_info.getJSON() ==
{"endpoint": "https://push...",
"keys":{"auth": "...", "p256dh": "..."}
}
)
This subscription_info block can be stored.
To send a subscription update:
.. code-block:: python
# Optional
# headers = py_vapid.sign({"aud": "http://your.site.com",
"sub": "mailto:your_admin@your.site.com"})
data = "Mary had a little lamb, with a nice mint jelly"
WebPusher(subscription_info).send(data, headers)
"""
subscription_info = {}
def __init__(self, subscription_info):
"""Initialize using the info provided by the client PushSubscription
object (See
https://developer.mozilla.org/en-US/docs/Web/API/PushManager/subscribe)
:param subscription_info: a dict containing the subscription_info from
the client.
"""
if 'endpoint' not in subscription_info:
raise WebPushException("subscription_info missing endpoint URL")
if 'keys' not in subscription_info:
raise WebPushException("subscription_info missing keys dictionary")
self.subscription_info = subscription_info
keys = self.subscription_info['keys']
for k in ['p256dh', 'auth']:
if keys.get(k) is None:
raise WebPushException("Missing keys value: %s", k)
receiver_raw = base64.urlsafe_b64decode(self._repad(keys['p256dh']))
if len(receiver_raw) != 65 and receiver_raw[0] != "\x04":
raise WebPushException("Invalid p256dh key specified")
self.receiver_key = receiver_raw
self.auth_key = base64.urlsafe_b64decode(self._repad(keys['auth']))
def _repad(self, str):
"""Add base64 padding to the end of a string, if required"""
return str + "===="[:len(str) % 4]
def encode(self, data):
"""Encrypt the data.
:param data: A serialized block of data (String, JSON, bit array,
etc.) Make sure that whatever you send, your client knows how
to understand it.
"""
# Salt is a random 16 byte array.
salt = os.urandom(16)
# The server key is an ephemeral ECDH key used only for this
# transaction
server_key = pyelliptic.ECC(curve="prime256v1")
# the ID is the base64 of the raw key, minus the leading "\x04"
# ID tag.
server_key_id = base64.urlsafe_b64encode(server_key.get_pubkey()[1:])
# http_ece requires that these both be set BEFORE encrypt or
# decrypt is called.
http_ece.keys[server_key_id] = server_key
http_ece.labels[server_key_id] = "P-256"
encrypted = http_ece.encrypt(
data,
salt=salt,
keyid=server_key_id,
dh=self.receiver_key,
authSecret=self.auth_key)
return {
'crypto_key': base64.urlsafe_b64encode(
server_key.get_pubkey()).strip('='),
'salt': base64.urlsafe_b64encode(salt).strip("="),
'body': encrypted,
}
def send(self, data, headers={}, ttl=0):
"""Encode and send the data to the Push Service.
:param data: A serialized block of data (see encode() ).
:param headers: A dictionary containing any additional HTTP headers.
:param ttl: The Time To Live in seconds for this message if the
recipient is not online. (Defaults to "0", which discards the
message immediately if the recipient is unavailable.)
"""
# Encode the data.
encoded = self.encode(data)
# Append the p256dh to the end of any existing crypto-key
crypto_key = headers.get("crypto-key", "")
if crypto_key:
crypto_key += ','
crypto_key += "keyid=p256dh;dh=" + encoded["crypto_key"]
headers.update({
'crypto-key': crypto_key,
'content-encoding': 'aesgcm',
'encryption': "keyid=p256dh;salt=" + encoded['salt'],
})
if 'ttl' not in headers or ttl:
headers['ttl'] = ttl
# Additionally useful headers:
# Authorization / Crypto-Key (VAPID headers)
return requests.post(self.subscription_info['endpoint'],
data=encoded.get('body'),
headers=headers)

117
webpush/publish.py
View File

@ -1,117 +0,0 @@
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.
import base64
import os
import http_ece
import pyelliptic
VAPID = True
try:
# from https://github.com/mozilla-services/vapid/tree/master/python
from py_vapid import Vapid
except ImportError:
VAPID = False
class WebPushException(Exception):
pass
class WebPusher:
def __init__(self, keys):
for k in ['p256dh', 'auth']:
if keys.get(k) is None:
raise WebPushException("Missing keys value: %s", k)
receiverRaw = base64.urlsafe_b64decode(self._repad(keys['p256dh']))
if len(receiverRaw) != 65 and receiverRaw[0] != "\x04":
raise WebPushException("Invalid p256dh key specified")
self.receiverKey = receiverRaw
self.authKey = base64.urlsafe_b64decode(self._repad(keys['auth']))
def _repad(self, str):
return str + "===="[:len(str) % 4]
def encode(self, data):
# Salt is a random 16 byte array.
salt = os.urandom(16)
# The server key is an ephemeral ECDH key used only for this
# transaction
serverKey = pyelliptic.ECC(curve="prime256v1")
# the ID is the base64 of the raw key, minus the leading "\x04"
# ID tag.
serverKeyID = base64.urlsafe_b64encode(serverKey.get_pubkey()[1:])
http_ece.keys[serverKeyID] = serverKey
http_ece.labels[serverKeyID] = "P-256"
encrypted = http_ece.encrypt(
data,
salt=salt,
keyid=serverKeyID,
dh=self.receiverKey,
authSecret=self.authKey)
return {
'cryptokey': base64.urlsafe_b64encode(
serverKey.get_pubkey()).strip('='),
'salt': base64.urlsafe_b64encode(salt).strip("="),
'body': encrypted,
}
def to_curl(self, endpoint, encode, headers={}, dataFile="encrypted.data"):
cryptokey = "keyid=p256dh;dh=%s" % encoded.get("cryptokey")
if headers.get('crypto-key'):
cryptokey = headers.get('crypto-key') + ','
headers["crypto-key"] = headers.get("crypto-key", "") + cryptokey
headers["TTL"] = 60
headers["content-encoding"] = "aesgcm"
headers["encryption"] = "keyid=p256dh;salt=%s" % encoded.get("salt")
reply = "curl -v -X POST %s " % endpoint
for key in headers:
reply += """-H "%s: %s" """ % (key, headers.get(key))
if dataFile:
reply += "--data-binary @%s" % dataFile
return reply
if __name__ == "__main__":
# The client provides the following values:
endpoint = ("https://updates.push.services.mozilla.com/push/v1/gAAAAABXAuZ"
"mKfEEyPbYfLXqtPW-yblFhEj-wjW5XHPJ3SMqjv9LlDWOAY9ljyZ80R4xHfD8"
"x2D_20j5mH4nbRQFyyCS33uyLgTp56zizeaitkMsw5EoAM8sRN_fz0Aaezrk9"
"W5uKpaf")
keys = {
"p256dh": ("BOrnIslXrUow2VAzKCUAE4sIbK00daEZCswOcf8m3T"
"F8V82B-OpOg5JbmYLg44kRcvQC1E2gMJshsUYA-_zMPR8"),
"auth": "k8JV6sjdbhAi1n3_LDBLvA",
}
# This is the optional VAPID data
vapid_claims = {
"aud": "http://example.com",
"sub": "mailto:admin@example.com",
}
data = "Mary had a little lamb, with a nice mint jelly."
vapid_headers = {}
if VAPID:
# You should only generate keys once, and write them out to
# safe storage. See https://github.com/mozilla-services/vapid for
# details.
vapid = Vapid()
vapid.generate_keys()
vapid_headers = vapid.sign(vapid_claims)
push = WebPusher(keys=keys)
encoded = push.encode(data)
with open("encrypted.data", "w") as out:
out.write(encoded.get('body'))
print push.to_curl(endpoint, encoded, vapid_headers)

108
webpush/tests/test_webpush.py Normal file
View File

@ -0,0 +1,108 @@
import base64
import os
import unittest
import http_ece
from mock import patch
from nose.tools import eq_, ok_
import pyelliptic
from webpush import WebPusher, WebPushException
class WebpushTestCase(unittest.TestCase):
def _gen_subscription_info(self, recv_key):
return {
"endpoint": "https://example.com/",
"keys": {
'auth': base64.urlsafe_b64encode(os.urandom(16)).strip('='),
'p256dh': base64.urlsafe_b64encode(
recv_key.get_pubkey()).strip('='),
}
}
def test_init(self):
# use static values so we know what to look for in the reply
subscription_info = {
"endpoint": "https://example.com/",
"keys": {
"p256dh": ("BOrnIslXrUow2VAzKCUAE4sIbK00daEZCswOcf8m3T"
"F8V82B-OpOg5JbmYLg44kRcvQC1E2gMJshsUYA-_zMPR8"),
"auth": "k8JV6sjdbhAi1n3_LDBLvA"
}
}
self.assertRaises(
WebPushException,
WebPusher,
{"keys": {'p256dh': 'AAA=', 'auth': 'AAA='}})
self.assertRaises(
WebPushException,
WebPusher,
{"endpoint": "https://example.com"})
self.assertRaises(
WebPushException,
WebPusher,
{"endpoint": "https://example.com", "keys": {'p256dh': 'AAA='}})
self.assertRaises(
WebPushException,
WebPusher,
{"endpoint": "https://example.com", "keys": {'auth': 'AAA='}})
self.assertRaises(
WebPushException,
WebPusher,
{"endpoint": "https://example.com",
"keys": {'p256dh': 'AAA=', 'auth': 'AAA='}})
push = WebPusher(subscription_info)
eq_(push.subscription_info, subscription_info)
eq_(push.receiver_key, ('\x04\xea\xe7"\xc9W\xadJ0\xd9P3(%\x00\x13\x8b'
'\x08l\xad4u\xa1\x19\n\xcc\x0eq\xff&\xdd1'
'|W\xcd\x81\xf8\xeaN\x83\x92[\x99\x82\xe0\xe3'
'\x89\x11r\xf4\x02\xd4M\xa00\x9b!\xb1F\x00'
'\xfb\xfc\xcc=\x1f'))
eq_(push.auth_key, '\x93\xc2U\xea\xc8\xddn\x10"\xd6}\xff,0K\xbc')
def test_encode(self):
recv_key = pyelliptic.ECC(curve="prime256v1")
subscription_info = self._gen_subscription_info(recv_key)
data = "Mary had a little lamb, with some nice mint jelly"
push = WebPusher(subscription_info)
encoded = push.encode(data)
keyid = base64.urlsafe_b64encode(recv_key.get_pubkey()[1:])
http_ece.keys[keyid] = recv_key
http_ece.labels[keyid] = 'P-256'
# Convert these b64 strings into their raw, binary form.
raw_salt = base64.urlsafe_b64decode(push._repad(encoded['salt']))
raw_dh = base64.urlsafe_b64decode(push._repad(encoded['crypto_key']))
raw_auth = base64.urlsafe_b64decode(
push._repad(subscription_info['keys']['auth']))
decoded = http_ece.decrypt(
buffer=encoded['body'],
salt=raw_salt,
dh=raw_dh,
keyid=keyid,
authSecret=raw_auth
)
eq_(decoded, data)
@patch("requests.post")
def test_send(self, mock_post):
recv_key = pyelliptic.ECC(curve="prime256v1")
subscription_info = self._gen_subscription_info(recv_key)
headers = {"crypto-key": "pre-existing",
"authentication": "bearer vapid"}
data = "Mary had a little lamb"
WebPusher(subscription_info).send(data, headers)
eq_(subscription_info.get('endpoint'), mock_post.call_args[0][0])
pheaders = mock_post.call_args[1].get('headers')
eq_(pheaders.get('ttl'), 0)
ok_('encryption' in pheaders)
eq_(pheaders.get('authentication'), headers.get('authentication'))
ckey = pheaders.get('crypto-key')
ok_('pre-existing,' in ckey)
eq_(pheaders.get('content-encoding'), 'aesgcm')