apns-server-http2.py

# -*- coding: utf-8 -*-
"""
apns-server-http2.py
~~~~~~~~~~~~~~~~~

A simple Python script to mock the Apple push notification gateway (requires Python 3.5).

Usage:
    $ python3.5 apns-server-http2.py /path/to/certificate.pem /path/to/key.pem certificate_password port

Example:
    $ python3.5 apns-server-http2.py cert.pem key.pem judepereira 8443

Generating the certificate:
    $ openssl req -x509 -newkey rsa:2048 -keyout key.pem -out cert.pem -days 300

Installing the dependencies:
    $ easy_install-3.5 asyncio h2


Note: If you specify the port as 443, you will need to run this script as root
(ports below 1024 cannot be opened by everybody)

This script has been adapted from http://python-hyper.org/h2/en/stable/asyncio-example.html
"""
import json
import ssl
import sys
from random import randint

import asyncio
import h2
from h2.connection import H2Connection
from h2.events import DataReceived, RequestReceived


class H2Protocol(asyncio.Protocol):
    def __init__(self):
        self.conn = H2Connection(client_side=False)
        self.transport = None

    def connection_made(self, transport: asyncio.Transport):
        self.transport = transport
        self.conn.initiate_connection()
        self.transport.write(self.conn.data_to_send())

    def data_received(self, data: bytes):
        events = self.conn.receive_data(data)
        # print("data_received")
        for event in events:
            # print("processing event {}", event)

            if isinstance(event, RequestReceived):
                self.request_received(event.stream_id)
            elif isinstance(event, DataReceived):
                self.transport.write(self.conn.data_to_send())

    def request_received(self, stream_id: int):
        print("request received on stream {}", stream_id)
        i = randint(0, 9)

        if i < 7:
            response_body = None
            status = "200"
        elif i == 7:
            response_body = {"reason": "BadDeviceToken"}
            status = "400"
        else:
            response_body = {"reason": "Unregistered"}
            status = "410"

        if response_body is not None:
            data = json.dumps(response_body).encode("utf8")
        else:
            data = "".encode("utf8")

        response_headers = (
            (':status', status),
            ('content-type', 'application/json'),
        )

        self.conn.send_headers(stream_id, response_headers)
        self.conn.send_data(stream_id, data, end_stream=True)


if len(sys.argv) != 5:
    print('usage: apns-server-http2.py /path/to/certificate.pem /path/to/key.pem certificate_password 8443')
    sys.exit(2)

h2.settings.HEADER_TABLE_SIZE = 1

cert = sys.argv[1]
key = sys.argv[2]
password = sys.argv[3]
port = sys.argv[4]

ssl_context = ssl.SSLContext(ssl.PROTOCOL_SSLv23)
ssl_context.options |= (
    ssl.OP_NO_SSLv2 | ssl.OP_NO_SSLv3
)
ssl_context.set_ciphers("ECDHE+AESGCM")
ssl_context.load_cert_chain(certfile=cert, keyfile=key, password=password)
ssl_context.set_alpn_protocols(["h2"])

loop = asyncio.get_event_loop()
# Each client connection will create a new protocol instance
coro = loop.create_server(H2Protocol, '127.0.0.1', port, ssl=ssl_context)
server = loop.run_until_complete(coro)

# Serve requests until Ctrl+C is pressed
print('Serving on {}. Hit ctrl+c to stop'.format(server.sockets[0].getsockname()))
try:
    loop.run_forever()
except KeyboardInterrupt:
    pass

server.close()
loop.run_until_complete(server.wait_closed())
loop.close()