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

feat: Allow VAPID with [gf]cm 

* added primative CLI
* dump as curl

closes #44
This commit is contained in:
jrconlin 2017-03-07 17:05:57 -08:00 committed by jr conlin
parent b6348a6dc4
commit caf331dab8
10 changed files with 601 additions and 108 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
MANIFEST.in
View File

@ -2,4 +2,4 @@ include *.md
include *.txt
include setup.*
include LICENSE
recursive-include pywebpush
recursive-include pywebpush *.py

139
README.md
View File

@ -1,20 +1,19 @@
[![Build_Status](https://travis-ci.org/jrconlin/pywebpush.svg?branch=master)](https://travis-ci.org/jrconlin/pywebpush)
[![Build_Status](https://travis-ci.org/web-push-libs/pywebpush.svg?branch=master)](https://travis-ci.org/web-push-libs/pywebpush)
[![Requirements
Status](https://requires.io/github/web-push-libs/pywebpush/requirements.svg?branch=master)]
Status](https://requires.io/github/web-push-libs/pywebpush/requirements.svg?branch=feat%2F44)](https://requires.io/github/web-push-libs/pywebpush/requirements/?branch=master)
# Webpush Data encryption library for Python
This is a work in progress.
This library is available on [pypi as
pywebpush](https://pypi.python.org/pypi/pywebpush).
Source is available on [github](https://github.com/jrconlin/pywebpush)
This library is available on [pypi as pywebpush](https://pypi.python.org/pypi/pywebpush).
Source is available on
[github](https://github.com/mozilla-services/pywebpush).
## Installation
You'll need to run `python virtualenv`.
Then
```
```commandline
bin/pip install -r requirements.txt
bin/python setup.py develop
```
@ -25,42 +24,128 @@ 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.
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:
```
As illustration, a `subscription_info` object may look like:
```json
{"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
a new push subscription update is left as an exercise 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."`)
### Sending Data using `webpush()` One Call
gcm_key is the API key obtained from the Google Developer Console.
It is only needed if endpoint is
https://android.googleapis.com/gcm/send
In many cases, your code will be sending a single message to many
recipients. There's a "One Call" function which will make things
easier.
`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.
```pythonstub
from pywebpush import webpush
to send:
webpush(subscription_info,
data,
vapid_private_key="Private Key or File Path[1]",
vapid_claims={"sub": "mailto:YourEmailAddress"})
```
WebPusher(subscription_info).send(data, headers)
This will encode `data`, add the appropriate VAPID auth headers if required and send it to the push server identified
in the `subscription_info` block.
**Parameters**
*subscription_info* - The `dict` of the subscription info (described above).
*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."`)
*vapid_claims* - a `dict` containing the VAPID claims required for authorization (See
[py_vapid](https://github.com/web-push-libs/vapid/tree/master/python) for more details)
*vapid_private_key* - Either a path to a VAPID EC2 private key PEM file, or a string containing the DER representation.
(See [py_vapid](https://github.com/web-push-libs/vapid/tree/master/python) for more details.) The `private_key` may be
a base64 encoded DER formatted private key, or the path to an OpenSSL exported private key file.
e.g. the output of:
```commandline
openssl ecparam -name prime256v1 -genkey -noout -out private_key.pem
```
to send for Chrome:
**Example**
```pythonstub
from pywebpush import webpush, WebPushException
try:
webpush(
subscription_info={
"endpoint": "https://push.example.com/v1/12345",
"keys": {
"p256dh": "0123abcde...",
"auth": "abc123..."
}},
data="Mary had a little lamb, with a nice mint jelly",
vapid_private_key="path/to/vapid_private.pem",
vapid_claims={
"sub": "YourNameHere@example.org",
}
)
except WebPushException as ex:
print("I'm sorry, Dave, but I can't do that: {}", repr(ex))
```
### Methods
If you expect to resend to the same recipient, or have more needs than just sending data quickly, you
can pass just `wp = WebPusher(subscription_info)`. This will return a `WebPusher` object.
The following methods are available:
#### `.send(data, headers={}, ttl=0, gcm_key="", reg_id="", content_encoding="aesgcm", curl=False)`
Send the data using additional parameters. On error, returns a `WebPushException`
**Parameters**
*data* Binary string of data to send
*headers* A `dict` containing any additional headers to send
*ttl* Message Time To Live on Push Server waiting for the client to reconnect (in seconds)
*gcm_key* Google Cloud Messaging key (if using the older GCM push system) This is the API key obtained from the Google
Developer Console.
*reg_id* Google Cloud Messaging registration ID (will be extracted from endpoint if not specified)
*content_encoding* ECE content encoding type (defaults to "aesgcm")
*curl* Do not execute the POST, but return as a `curl` command. This will write the encrypted content to a local file
named `encrpypted.data`. This command is meant to be used for debugging purposes.
**Example**
to send from Chrome using the old GCM mode:
```pythonstub
WebPusher(subscription_info).send(data, headers, ttl, gcm_key)
```
You can also simply encode the data to send later by calling
#### `.encode(data, content_encoding="aesgcm")`
Encode the `data` for future use. On error, returns a `WebPushException`
**Parameters**
*data* Binary string of data to send
*content_encoding* ECE content encoding type (defaults to "aesgcm")
**Example**
```pythonstub
encoded_data = WebPush(subscription_info).encode(data)
```
encoded = WebPush(subscription_info).encode(data)
```

149
README.rst
View File

@ -1,18 +1,18 @@
|Build\_Status| [|Requirements Status|]
|Build\_Status| |Requirements Status|
Webpush Data encryption library for Python
==========================================
This is a work in progress. This library is available on `pypi as
pywebpush <https://pypi.python.org/pypi/pywebpush>`__. Source is
available on `github <https://github.com/web-push-libs/pywebpush>`__
available on `github <https://github.com/mozilla-services/pywebpush>`__.
Installation
------------
You'll need to run ``python virtualenv``. Then
::
.. code:: commandline
bin/pip install -r requirements.txt
bin/python setup.py develop
@ -27,45 +27,150 @@ returns a
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:
As illustration, a ``subscription_info`` object may look like:
::
.. code:: json
{"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.
referenced to the user who requested it, and recall it when there's a
new push subscription update is left as an exercise for the reader.
The data can be any serial content (string, bit array, serialized JSON,
Sending Data using ``webpush()`` One Call
~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~~
In many cases, your code will be sending a single message to many
recipients. There's a "One Call" function which will make things easier.
.. code:: pythonstub
from pywebpush import webpush
webpush(subscription_info,
data,
vapid_private_key="Private Key or File Path[1]",
vapid_claims={"sub": "mailto:YourEmailAddress"})
This will encode ``data``, add the appropriate VAPID auth headers if
required and send it to the push server identified in the
``subscription_info`` block.
**Parameters**
*subscription\_info* - The ``dict`` of the subscription info (described
above).
*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."``)
gcm\_key is the API key obtained from the Google Developer Console. It
is only needed if endpoint is https://android.googleapis.com/gcm/send
*vapid\_claims* - a ``dict`` containing the VAPID claims required for
authorization (See
`py\_vapid <https://github.com/web-push-libs/vapid/tree/master/python>`__
for more details)
``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.
*vapid\_private\_key* - Either a path to a VAPID EC2 private key PEM
file, or a string containing the DER representation. (See
`py\_vapid <https://github.com/web-push-libs/vapid/tree/master/python>`__
for more details.) The ``private_key`` may be a base64 encoded DER
formatted private key, or the path to an OpenSSL exported private key
file.
to send:
e.g. the output of:
::
.. code:: commandline
WebPusher(subscription_info).send(data, headers)
openssl ecparam -name prime256v1 -genkey -noout -out private_key.pem
to send for Chrome:
**Example**
::
.. code:: pythonstub
from pywebpush import webpush, WebPushException
try:
webpush(
subscription_info={
"endpoint": "https://push.example.com/v1/12345",
"keys": {
"p256dh": "0123abcde...",
"auth": "abc123..."
}},
data="Mary had a little lamb, with a nice mint jelly",
vapid_private_key="path/to/vapid_private.pem",
vapid_claims={
"sub": "YourNameHere@example.org",
}
)
except WebPushException as ex:
print("I'm sorry, Dave, but I can't do that: {}", repr(ex))
Methods
~~~~~~~
If you expect to resend to the same recipient, or have more needs than
just sending data quickly, you can pass just
``wp = WebPusher(subscription_info)``. This will return a ``WebPusher``
object.
The following methods are available:
``.send(data, headers={}, ttl=0, gcm_key="", reg_id="", content_encoding="aesgcm", curl=False)``
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
Send the data using additional parameters. On error, returns a
``WebPushException``
**Parameters**
*data* Binary string of data to send
*headers* A ``dict`` containing any additional headers to send
*ttl* Message Time To Live on Push Server waiting for the client to
reconnect (in seconds)
*gcm\_key* Google Cloud Messaging key (if using the older GCM push
system) This is the API key obtained from the Google Developer Console.
*reg\_id* Google Cloud Messaging registration ID (will be extracted from
endpoint if not specified)
*content\_encoding* ECE content encoding type (defaults to "aesgcm")
*curl* Do not execute the POST, but return as a ``curl`` command. This
will write the encrypted content to a local file named
``encrpypted.data``. This command is meant to be used for debugging
purposes.
**Example**
to send from Chrome using the old GCM mode:
.. code:: pythonstub
WebPusher(subscription_info).send(data, headers, ttl, gcm_key)
You can also simply encode the data to send later by calling
``.encode(data, content_encoding="aesgcm")``
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
::
Encode the ``data`` for future use. On error, returns a
``WebPushException``
encoded = WebPush(subscription_info).encode(data)
**Parameters**
*data* Binary string of data to send
*content\_encoding* ECE content encoding type (defaults to "aesgcm")
**Example**
.. code:: pythonstub
encoded_data = WebPush(subscription_info).encode(data)
.. |Build\_Status| image:: https://travis-ci.org/web-push-libs/pywebpush.svg?branch=master
:target: https://travis-ci.org/web-push-libs/pywebpush
.. |Requirements Status| image:: https://requires.io/github/web-push-libs/pywebpush/requirements.svg?branch=master
.. |Requirements Status| image:: https://requires.io/github/web-push-libs/pywebpush/requirements.svg?branch=feat%2F44
:target: https://requires.io/github/web-push-libs/pywebpush/requirements/?branch=master

2
convert_readme.sh
View File

@ -1 +1 @@
pandoc --from=markdown --to=rst --output README.rst README.md
pandoc --from=markdown --to=rst --output README.rst README.md

180
pywebpush/__init__.py
View File

@ -6,10 +6,16 @@ import base64
import json
import os
try:
from urllib.parse import urlparse
except ImportError: # pragma nocover
from urlparse import urlparse
import six
import http_ece
import pyelliptic
import requests
from py_vapid import Vapid
class WebPushException(Exception):
@ -64,7 +70,7 @@ class WebPusher:
(e.g.
.. code-block:: javascript
subscription_info.getJSON() ==
{"endpoint": "https://push...",
{"endpoint": "https://push.server.com/...",
"keys":{"auth": "...", "p256dh": "..."}
}
)
@ -75,7 +81,7 @@ class WebPusher:
.. code-block:: python
# Optional
# headers = py_vapid.sign({"aud": "http://your.site.com",
# headers = py_vapid.sign({"aud": "https://push.server.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)
@ -95,24 +101,27 @@ class WebPusher:
:param subscription_info: a dict containing the subscription_info from
the client.
:type subscription_info: dict
"""
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)
if isinstance(keys[k], six.string_types):
keys[k] = bytes(keys[k].encode('utf8'))
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']))
self.auth_key = self.receiver_key = None
if 'keys' in 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)
if isinstance(keys[k], six.string_types):
keys[k] = bytes(keys[k].encode('utf8'))
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, data):
"""Add base64 padding to the end of a string, if required"""
@ -133,6 +142,10 @@ class WebPusher:
"""
# Salt is a random 16 byte array.
if not data:
return
if not self.auth_key or not self.receiver_key:
raise WebPushException("No keys specified in subscription info")
salt = None
if content_encoding not in self.valid_encodings:
raise WebPushException("Invalid content encoding specified. "
@ -173,19 +186,58 @@ class WebPusher:
reply['salt'] = base64.urlsafe_b64encode(salt).strip(b'=')
return reply
def as_curl(self, endpoint, encoded_data, headers):
"""Return the send as a curl command.
Useful for debugging. This will write out the encoded data to a local
file named `encrypted.data`
:param endpoint: Push service endpoint URL
:type endpoint: basestring
:param encoded_data: byte array of encoded data
:type encoded_data: bytearray
:param headers: Additional headers for the send
:type headers: dict
:returns string
"""
header_list = [
'-H "{}: {}" \\ \n'.format(
key.lower(), val) for key, val in headers.items()
]
data = ""
if encoded_data:
with open("encrypted.data", "wb") as f:
f.write(encoded_data)
data = "--data-binary @encrypted.data"
if 'content-length' not in headers:
header_list.append(
'-H "content-length: {}" \\ \n'.format(len(data)))
return ("""curl -vX POST {url} \\\n{headers}{data}""".format(
url=endpoint, headers="".join(header_list), data=data))
def send(self, data=None, headers=None, ttl=0, gcm_key=None, reg_id=None,
content_encoding="aesgcm"):
content_encoding="aesgcm", curl=False):
"""Encode and send the data to the Push Service.
:param data: A serialized block of data (see encode() ).
:type data: str
:param headers: A dictionary containing any additional HTTP headers.
:type headers: dict
: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.)
:type ttl: int
:param gcm_key: API key obtained from the Google Developer Console.
Needed if endpoint is https://android.googleapis.com/gcm/send
:type gcm_key: string
:param reg_id: registration id of the recipient. If not provided,
it will be extracted from the endpoint.
:type reg_id: str
:param content_encoding: ECE content encoding (defaults to "aesgcm")
:type content_encoding: str
:param curl: Display output as `curl` command instead of sending
:type curl: bool
"""
# Encode the data.
@ -206,15 +258,12 @@ class WebPusher:
"keyid=p256dh;dh=" + encoded["crypto_key"].decode('utf8'))
headers.update({
'crypto-key': crypto_key,
'content-encoding': 'aesgcm',
'content-encoding': content_encoding,
'encryption': "keyid=p256dh;salt=" +
encoded['salt'].decode('utf8'),
})
gcm_endpoint = 'https://android.googleapis.com/gcm/send'
if self.subscription_info['endpoint'].startswith(gcm_endpoint):
if not gcm_key:
raise WebPushException("API key not provided for gcm endpoint")
endpoint = gcm_endpoint
if gcm_key:
endpoint = 'https://android.googleapis.com/gcm/send'
reg_ids = []
if not reg_id:
reg_id = self.subscription_info['endpoint'].rsplit('/', 1)[-1]
@ -239,15 +288,80 @@ class WebPusher:
headers['ttl'] = str(ttl or 0)
# Additionally useful headers:
# Authorization / Crypto-Key (VAPID headers)
return self._post(endpoint, encoded_data, headers)
def _post(self, url, data, headers):
"""Make POST request on specified Web Push endpoint with given data
and headers.
Subclass this class and override this method if you want to use any
other networking library or interface and keep the business logic.
"""
return requests.post(url,
data=data,
if curl:
return self.as_curl(endpoint, encoded_data, headers)
return requests.post(endpoint,
data=encoded_data,
headers=headers)
def webpush(subscription_info,
data=None,
vapid_private_key=None,
vapid_claims=None,
content_encoding="aesgcm",
curl=False):
"""
One call solution to endcode and send `data` to the endpoint
contained in `subscription_info` using optional VAPID auth headers.
in example:
.. code-block:: python
from pywebpush import python
webpush(
subscription_info={
"endpoint": "https://push.example.com/v1/abcd",
"keys": {"p256dh": "0123abcd...",
"auth": "001122..."}
},
data="Mary had a little lamb, with a nice mint jelly",
vapid_private_key="path/to/key.pem",
vapid_claims={"sub": "YourNameHere@example.com"}
)
No additional method call is required. Any non-success will throw a
`WebPushException`.
:param subscription_info: Provided by the client call
:type subscription_info: dict
:param data: Serialized data to send
:type data: str
:param vapid_private_key: Dath to vapid private key PEM or encoded str
:type vapid_private_key: str
:param vapid_claims: Dictionary of claims ('sub' required)
:type vapid_claims: dict
:param content_encoding: Optional content type string
:type content_encoding: str
:param curl: Return as "curl" string instead of sending
:type curl: bool
:return requests.Response or string
"""
vapid_headers = None
if vapid_claims:
if not vapid_claims.get('aud'):
url = urlparse(subscription_info.get('endpoint'))
aud = "{}://{}/".format(url.scheme, url.netloc)
vapid_claims['aud'] = aud
if not vapid_private_key:
raise WebPushException("VAPID dict missing 'private_key'")
if os.path.isfile(vapid_private_key):
# Presume that key from file is handled correctly by
# py_vapid.
vv = Vapid(private_key_file=vapid_private_key) # pragma no cover
else:
vv = Vapid(private_key=vapid_private_key)
vapid_headers = vv.sign(vapid_claims)
result = WebPusher(subscription_info).send(
data,
vapid_headers,
content_encoding=content_encoding,
curl=curl,
)
if not curl and result.status_code > 202:
raise WebPushException("Push failed: {}:".format(
result, result.text))
return result

63
pywebpush/__main__.py Normal file
View File

@ -0,0 +1,63 @@
import argparse
import os
import json
from pywebpush import webpush
def get_config():
parser = argparse.ArgumentParser(description="WebPush tool")
parser.add_argument("--data", '-d', help="Data file")
parser.add_argument("--info", "-i", help="Subscription Info JSON file")
parser.add_argument("--head", help="Header Info JSON file")
parser.add_argument("--claims", help="Vapid claim file")
parser.add_argument("--key", help="Vapid private key file path")
parser.add_argument("--curl", help="Don't send, display as curl command",
default=False, action="store_true")
parser.add_argument("--encoding", default="aesgcm")
args = parser.parse_args()
if not args.info:
raise Exception("Subscription Info argument missing.")
if not os.path.exists(args.info):
raise Exception("Subscription Info file missing.")
try:
with open(args.info) as r:
args.sub_info = json.loads(r.read())
if args.data:
with open(args.data) as r:
args.data = r.read()
if args.head:
with open(args.head) as r:
args.head = json.loads(r.read())
if args.claims:
if not args.key:
raise Exception("No private --key specified for claims")
with open(args.claims) as r:
args.claims = json.loads(r.read())
except Exception as ex:
print("Couldn't read input {}.".format(ex))
raise ex
return args
def main():
""" Send data """
try:
args = get_config()
result = webpush(
args.sub_info,
data=args.data,
vapid_private_key=args.key,
vapid_claims=args.claims,
curl=args.curl,
content_encoding=args.encoding)
print(result)
except Exception as ex:
print("ERROR: {}".format(ex))
if __name__ == "__main__":
main()

147
pywebpush/tests/test_webpush.py
View File

@ -3,16 +3,26 @@ import json
import os
import unittest
from mock import patch
from nose.tools import eq_, ok_
from mock import patch, Mock
from nose.tools import eq_, ok_, assert_raises
import http_ece
import pyelliptic
from pywebpush import WebPusher, WebPushException, CaseInsensitiveDict
from pywebpush import WebPusher, WebPushException, CaseInsensitiveDict, webpush
class WebpushTestCase(unittest.TestCase):
def _gen_subscription_info(self, recv_key, endpoint="https://example.com"):
# This is a exported DER formatted string of an ECDH public key
# This was lifted from the py_vapid tests.
vapid_key = (
"MHcCAQEEIPeN1iAipHbt8+/KZ2NIF8NeN24jqAmnMLFZEMocY8RboAoGCCqGSM49"
"AwEHoUQDQgAEEJwJZq/GN8jJbo1GGpyU70hmP2hbWAUpQFKDByKB81yldJ9GTklB"
"M5xqEwuPM7VuQcyiLDhvovthPIXx+gsQRQ=="
)
def _gen_subscription_info(self, recv_key,
endpoint="https://example.com/"):
return {
"endpoint": endpoint,
"keys": {
@ -41,10 +51,6 @@ class WebpushTestCase(unittest.TestCase):
WebPushException,
WebPusher,
{"keys": {'p256dh': 'AAA=', 'auth': 'AAA='}})
self.assertRaises(
WebPushException,
WebPusher,
{"endpoint": "https://example.com"})
self.assertRaises(
WebPushException,
WebPusher,
@ -121,14 +127,84 @@ class WebpushTestCase(unittest.TestCase):
ok_('pre-existing' in ckey)
eq_(pheaders.get('content-encoding'), 'aesgcm')
@patch("requests.post")
def test_send_vapid(self, mock_post):
mock_post.return_value = Mock()
mock_post.return_value.status_code = 200
recv_key = pyelliptic.ECC(curve="prime256v1")
subscription_info = self._gen_subscription_info(recv_key)
data = "Mary had a little lamb"
webpush(
subscription_info=subscription_info,
data=data,
vapid_private_key=self.vapid_key,
vapid_claims={"sub": "mailto:ops@example.com"}
)
eq_(subscription_info.get('endpoint'), mock_post.call_args[0][0])
pheaders = mock_post.call_args[1].get('headers')
eq_(pheaders.get('ttl'), '0')
def repad(str):
return str + "===="[:len(str) % 4]
auth = json.loads(
base64.urlsafe_b64decode(
repad(pheaders['authorization'].split('.')[1])
).decode('utf8')
)
ok_(subscription_info.get('endpoint').startswith(auth['aud']))
ok_('encryption' in pheaders)
ok_('WebPush' in pheaders.get('authorization'))
ckey = pheaders.get('crypto-key')
ok_('p256ecdsa=' in ckey)
ok_('dh=' in ckey)
eq_(pheaders.get('content-encoding'), 'aesgcm')
@patch("requests.post")
def test_send_bad_vapid_no_key(self, mock_post):
mock_post.return_value = Mock()
mock_post.return_value.status_code = 200
recv_key = pyelliptic.ECC(curve="prime256v1")
subscription_info = self._gen_subscription_info(recv_key)
data = "Mary had a little lamb"
assert_raises(WebPushException,
webpush,
subscription_info=subscription_info,
data=data,
vapid_claims={
"aud": "https://example.com",
"sub": "mailto:ops@example.com"
}
)
@patch("requests.post")
def test_send_bad_vapid_bad_return(self, mock_post):
mock_post.return_value = Mock()
mock_post.return_value.status_code = 410
recv_key = pyelliptic.ECC(curve="prime256v1")
subscription_info = self._gen_subscription_info(recv_key)
data = "Mary had a little lamb"
assert_raises(WebPushException,
webpush,
subscription_info=subscription_info,
data=data,
vapid_claims={
"aud": "https://example.com",
"sub": "mailto:ops@example.com"
},
vapid_private_key=self.vapid_key
)
@patch("requests.post")
def test_send_empty(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 = None
WebPusher(subscription_info).send(data, headers)
WebPusher(subscription_info).send('', 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')
@ -137,6 +213,27 @@ class WebpushTestCase(unittest.TestCase):
ckey = pheaders.get('crypto-key')
ok_('pre-existing' in ckey)
def test_encode_empty(self):
recv_key = pyelliptic.ECC(curve="prime256v1")
subscription_info = self._gen_subscription_info(recv_key)
headers = {"Crypto-Key": "pre-existing",
"Authentication": "bearer vapid"}
encoded = WebPusher(subscription_info).encode('', headers)
eq_(encoded, None)
def test_encode_no_crypto(self):
recv_key = pyelliptic.ECC(curve="prime256v1")
subscription_info = self._gen_subscription_info(recv_key)
del(subscription_info['keys'])
headers = {"Crypto-Key": "pre-existing",
"Authentication": "bearer vapid"}
data = 'Something'
pusher = WebPusher(subscription_info)
assert_raises(WebPushException,
pusher.encode,
data,
headers)
@patch("requests.post")
def test_send_no_headers(self, mock_post):
recv_key = pyelliptic.ECC(curve="prime256v1")
@ -149,6 +246,31 @@ class WebpushTestCase(unittest.TestCase):
ok_('encryption' in pheaders)
eq_(pheaders.get('content-encoding'), 'aesgcm')
@patch("pywebpush.open")
def test_as_curl(self, opener):
recv_key = pyelliptic.ECC(curve="prime256v1")
subscription_info = self._gen_subscription_info(recv_key)
result = webpush(
subscription_info,
data="Mary had a little lamb",
vapid_claims={
"aud": "https://example.com",
"sub": "mailto:ops@example.com"
},
vapid_private_key=self.vapid_key,
curl=True
)
for s in [
"curl -vX POST https://example.com",
"-H \"crypto-key: p256ecdsa=",
"-H \"content-encoding: aesgcm\"",
"-H \"authorization: WebPush ",
"-H \"encryption: keyid=p256dh;salt=",
"-H \"ttl: 0\"",
"-H \"content-length:"
]:
ok_(s in result)
def test_ci_dict(self):
ci = CaseInsensitiveDict({"Foo": "apple", "bar": "banana"})
eq_('apple', ci["foo"])
@ -167,11 +289,6 @@ class WebpushTestCase(unittest.TestCase):
"Authentication": "bearer vapid"}
data = "Mary had a little lamb"
wp = WebPusher(subscription_info)
self.assertRaises(
WebPushException,
wp.send,
data,
headers)
wp.send(data, headers, gcm_key="gcm_key_value")
pdata = json.loads(mock_post.call_args[1].get('data'))
pheaders = mock_post.call_args[1].get('headers')

8
requirements.txt
View File

@ -1,4 +1,4 @@
http-ece==0.7.0
python-jose>1.2.0
requests>2.11.0
flake8
http-ece==0.7.1
python-jose==1.3.2
requests==2.13.0
py-vapid==0.8.1

11
setup.py
View File

@ -3,7 +3,7 @@ import os
from setuptools import find_packages, setup
__version__ = "0.7.0"
__version__ = "0.8.0"
def read_from(file):
@ -13,6 +13,9 @@ def read_from(file):
l = l.strip()
if not l:
break
if l[:2] == '-r':
reply += read_from(l.split(' ')[1])
continue
if l[0] != '#' or l[:2] != '//':
reply.append(l)
return reply
@ -50,5 +53,9 @@ setup(
include_package_data=True,
zip_safe=False,
install_requires=read_from('requirements.txt'),
tests_require=read_from('test-requirements.txt')
tests_require=read_from('test-requirements.txt'),
entry_points="""
[console_scripts]
pywebpush = pywebpush.__main__:main
""",
)

8
test-requirements.txt
View File

@ -1,3 +1,5 @@
nose
coverage
mock>=1.0.1
-r requirements.txt
nose>=1.3.7
coverage>=4.3.4
mock==2.0.0
flake8