1656 lines
58 KiB
Cython
1656 lines
58 KiB
Cython
# nghttp2 - HTTP/2 C Library
|
|
|
|
# Copyright (c) 2013 Tatsuhiro Tsujikawa
|
|
|
|
# Permission is hereby granted, free of charge, to any person obtaining
|
|
# a copy of this software and associated documentation files (the
|
|
# "Software"), to deal in the Software without restriction, including
|
|
# without limitation the rights to use, copy, modify, merge, publish,
|
|
# distribute, sublicense, and/or sell copies of the Software, and to
|
|
# permit persons to whom the Software is furnished to do so, subject to
|
|
# the following conditions:
|
|
|
|
# The above copyright notice and this permission notice shall be
|
|
# included in all copies or substantial portions of the Software.
|
|
|
|
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
|
|
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
|
|
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
|
|
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
|
|
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
|
|
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
|
|
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
|
|
cimport cnghttp2
|
|
|
|
from libc.stdlib cimport malloc, free
|
|
from libc.string cimport memcpy, memset
|
|
from libc.stdint cimport uint8_t, uint16_t, uint32_t, int32_t
|
|
import logging
|
|
|
|
|
|
DEFAULT_HEADER_TABLE_SIZE = cnghttp2.NGHTTP2_DEFAULT_HEADER_TABLE_SIZE
|
|
DEFLATE_MAX_HEADER_TABLE_SIZE = 4096
|
|
|
|
HD_ENTRY_OVERHEAD = 32
|
|
|
|
class HDTableEntry:
|
|
|
|
def __init__(self, name, namelen, value, valuelen):
|
|
self.name = name
|
|
self.namelen = namelen
|
|
self.value = value
|
|
self.valuelen = valuelen
|
|
|
|
def space(self):
|
|
return self.namelen + self.valuelen + HD_ENTRY_OVERHEAD
|
|
|
|
cdef _get_pybytes(uint8_t *b, uint16_t blen):
|
|
return b[:blen]
|
|
|
|
cdef class HDDeflater:
|
|
'''Performs header compression. The constructor takes
|
|
|hd_table_bufsize_max| parameter, which limits the usage of header
|
|
table in the given amount of bytes. This is necessary because the
|
|
header compressor and decompressor share the same amount of
|
|
header table and the decompressor decides that number. The
|
|
compressor may not want to use all header table size because of
|
|
limited memory availability. In that case, the
|
|
|hd_table_bufsize_max| can be used to cap the upper limit of table
|
|
size whatever the header table size is chosen by the decompressor.
|
|
The default value of |hd_table_bufsize_max| is 4096 bytes.
|
|
|
|
The following example shows how to compress request header sets:
|
|
|
|
import binascii, nghttp2
|
|
|
|
deflater = nghttp2.HDDeflater()
|
|
res = deflater.deflate([(b'foo', b'bar'),
|
|
(b'baz', b'buz')])
|
|
print(binascii.b2a_hex(res))
|
|
|
|
'''
|
|
|
|
cdef cnghttp2.nghttp2_hd_deflater *_deflater
|
|
|
|
def __cinit__(self, hd_table_bufsize_max = DEFLATE_MAX_HEADER_TABLE_SIZE):
|
|
rv = cnghttp2.nghttp2_hd_deflate_new(&self._deflater,
|
|
hd_table_bufsize_max)
|
|
if rv != 0:
|
|
raise Exception(_strerror(rv))
|
|
|
|
def __dealloc__(self):
|
|
cnghttp2.nghttp2_hd_deflate_del(self._deflater)
|
|
|
|
def deflate(self, headers):
|
|
'''Compresses the |headers|. The |headers| must be sequence of tuple
|
|
of name/value pair, which are sequence of bytes (not unicode
|
|
string).
|
|
|
|
This function returns the encoded header block in byte string.
|
|
An exception will be raised on error.
|
|
|
|
'''
|
|
cdef cnghttp2.nghttp2_nv *nva = <cnghttp2.nghttp2_nv*>\
|
|
malloc(sizeof(cnghttp2.nghttp2_nv)*\
|
|
len(headers))
|
|
cdef cnghttp2.nghttp2_nv *nvap = nva
|
|
|
|
for k, v in headers:
|
|
nvap[0].name = k
|
|
nvap[0].namelen = len(k)
|
|
nvap[0].value = v
|
|
nvap[0].valuelen = len(v)
|
|
nvap[0].flags = cnghttp2.NGHTTP2_NV_FLAG_NONE
|
|
nvap += 1
|
|
|
|
cdef size_t outcap = 0
|
|
cdef ssize_t rv
|
|
cdef uint8_t *out
|
|
cdef size_t outlen
|
|
|
|
outlen = cnghttp2.nghttp2_hd_deflate_bound(self._deflater,
|
|
nva, len(headers))
|
|
|
|
out = <uint8_t*>malloc(outlen)
|
|
|
|
rv = cnghttp2.nghttp2_hd_deflate_hd(self._deflater, out, outlen,
|
|
nva, len(headers))
|
|
free(nva)
|
|
|
|
if rv < 0:
|
|
free(out)
|
|
|
|
raise Exception(_strerror(rv))
|
|
|
|
cdef bytes res
|
|
|
|
try:
|
|
res = out[:rv]
|
|
finally:
|
|
free(out)
|
|
|
|
return res
|
|
|
|
def change_table_size(self, hd_table_bufsize_max):
|
|
'''Changes header table size to |hd_table_bufsize_max| byte.
|
|
|
|
An exception will be raised on error.
|
|
|
|
'''
|
|
cdef int rv
|
|
rv = cnghttp2.nghttp2_hd_deflate_change_table_size(self._deflater,
|
|
hd_table_bufsize_max)
|
|
if rv != 0:
|
|
raise Exception(_strerror(rv))
|
|
|
|
def get_hd_table(self):
|
|
'''Returns copy of current dynamic header table.'''
|
|
cdef size_t length = cnghttp2.nghttp2_hd_deflate_get_num_table_entries(
|
|
self._deflater)
|
|
cdef const cnghttp2.nghttp2_nv *nv
|
|
res = []
|
|
for i in range(62, length + 1):
|
|
nv = cnghttp2.nghttp2_hd_deflate_get_table_entry(self._deflater, i)
|
|
k = _get_pybytes(nv.name, nv.namelen)
|
|
v = _get_pybytes(nv.value, nv.valuelen)
|
|
res.append(HDTableEntry(k, nv.namelen, v, nv.valuelen))
|
|
return res
|
|
|
|
cdef class HDInflater:
|
|
'''Performs header decompression.
|
|
|
|
The following example shows how to compress request header sets:
|
|
|
|
data = b'0082c5ad82bd0f000362617a0362757a'
|
|
inflater = nghttp2.HDInflater()
|
|
hdrs = inflater.inflate(data)
|
|
print(hdrs)
|
|
|
|
'''
|
|
|
|
cdef cnghttp2.nghttp2_hd_inflater *_inflater
|
|
|
|
def __cinit__(self):
|
|
rv = cnghttp2.nghttp2_hd_inflate_new(&self._inflater)
|
|
if rv != 0:
|
|
raise Exception(_strerror(rv))
|
|
|
|
def __dealloc__(self):
|
|
cnghttp2.nghttp2_hd_inflate_del(self._inflater)
|
|
|
|
def inflate(self, data):
|
|
'''Decompresses the compressed header block |data|. The |data| must be
|
|
byte string (not unicode string).
|
|
|
|
'''
|
|
cdef cnghttp2.nghttp2_nv nv
|
|
cdef int inflate_flags
|
|
cdef ssize_t rv
|
|
cdef uint8_t *buf = data
|
|
cdef size_t buflen = len(data)
|
|
res = []
|
|
while True:
|
|
inflate_flags = 0
|
|
rv = cnghttp2.nghttp2_hd_inflate_hd2(self._inflater, &nv,
|
|
&inflate_flags,
|
|
buf, buflen, 1)
|
|
if rv < 0:
|
|
raise Exception(_strerror(rv))
|
|
buf += rv
|
|
buflen -= rv
|
|
if inflate_flags & cnghttp2.NGHTTP2_HD_INFLATE_EMIT:
|
|
# may throw
|
|
res.append((nv.name[:nv.namelen], nv.value[:nv.valuelen]))
|
|
if inflate_flags & cnghttp2.NGHTTP2_HD_INFLATE_FINAL:
|
|
break
|
|
|
|
cnghttp2.nghttp2_hd_inflate_end_headers(self._inflater)
|
|
return res
|
|
|
|
def change_table_size(self, hd_table_bufsize_max):
|
|
'''Changes header table size to |hd_table_bufsize_max| byte.
|
|
|
|
An exception will be raised on error.
|
|
|
|
'''
|
|
cdef int rv
|
|
rv = cnghttp2.nghttp2_hd_inflate_change_table_size(self._inflater,
|
|
hd_table_bufsize_max)
|
|
if rv != 0:
|
|
raise Exception(_strerror(rv))
|
|
|
|
def get_hd_table(self):
|
|
'''Returns copy of current dynamic header table.'''
|
|
cdef size_t length = cnghttp2.nghttp2_hd_inflate_get_num_table_entries(
|
|
self._inflater)
|
|
cdef const cnghttp2.nghttp2_nv *nv
|
|
res = []
|
|
for i in range(62, length + 1):
|
|
nv = cnghttp2.nghttp2_hd_inflate_get_table_entry(self._inflater, i)
|
|
k = _get_pybytes(nv.name, nv.namelen)
|
|
v = _get_pybytes(nv.value, nv.valuelen)
|
|
res.append(HDTableEntry(k, nv.namelen, v, nv.valuelen))
|
|
return res
|
|
|
|
cdef _strerror(int liberror_code):
|
|
return cnghttp2.nghttp2_strerror(liberror_code).decode('utf-8')
|
|
|
|
def print_hd_table(hdtable):
|
|
'''Convenient function to print |hdtable| to the standard output. This
|
|
function does not work if header name/value cannot be decoded using
|
|
UTF-8 encoding.
|
|
|
|
s=N means the entry occupies N bytes in header table.
|
|
|
|
'''
|
|
idx = 0
|
|
for entry in hdtable:
|
|
idx += 1
|
|
print('[{}] (s={}) {}: {}'\
|
|
.format(idx, entry.space(),
|
|
entry.name.decode('utf-8'),
|
|
entry.value.decode('utf-8')))
|
|
|
|
try:
|
|
import socket
|
|
import io
|
|
import asyncio
|
|
import traceback
|
|
import sys
|
|
import email.utils
|
|
import datetime
|
|
import time
|
|
import ssl as tls
|
|
from urllib.parse import urlparse
|
|
except ImportError:
|
|
asyncio = None
|
|
|
|
# body generator flags
|
|
DATA_OK = 0
|
|
DATA_EOF = 1
|
|
DATA_DEFERRED = 2
|
|
|
|
class _ByteIOWrapper:
|
|
|
|
def __init__(self, b):
|
|
self.b = b
|
|
|
|
def generate(self, n):
|
|
data = self.b.read1(n)
|
|
if not data:
|
|
return None, DATA_EOF
|
|
return data, DATA_OK
|
|
|
|
def wrap_body(body):
|
|
if body is None:
|
|
return body
|
|
elif isinstance(body, str):
|
|
return _ByteIOWrapper(io.BytesIO(body.encode('utf-8'))).generate
|
|
elif isinstance(body, bytes):
|
|
return _ByteIOWrapper(io.BytesIO(body)).generate
|
|
elif isinstance(body, io.IOBase):
|
|
return _ByteIOWrapper(body).generate
|
|
else:
|
|
# assume that callable in the form f(n) returning tuple byte
|
|
# string and flag.
|
|
return body
|
|
|
|
def negotiated_protocol(ssl_obj):
|
|
protocol = ssl_obj.selected_alpn_protocol()
|
|
if protocol:
|
|
logging.info('alpn, protocol:%s', protocol)
|
|
return protocol
|
|
|
|
protocol = ssl_obj.selected_npn_protocol()
|
|
if protocol:
|
|
logging.info('npn, protocol:%s', protocol)
|
|
return protocol
|
|
|
|
return None
|
|
|
|
def set_application_protocol(ssl_ctx):
|
|
app_protos = [cnghttp2.NGHTTP2_PROTO_VERSION_ID.decode('utf-8')]
|
|
ssl_ctx.set_npn_protocols(app_protos)
|
|
if tls.HAS_ALPN:
|
|
ssl_ctx.set_alpn_protocols(app_protos)
|
|
|
|
cdef _get_stream_user_data(cnghttp2.nghttp2_session *session,
|
|
int32_t stream_id):
|
|
cdef void *stream_user_data
|
|
|
|
stream_user_data = cnghttp2.nghttp2_session_get_stream_user_data\
|
|
(session, stream_id)
|
|
if stream_user_data == NULL:
|
|
return None
|
|
|
|
return <object>stream_user_data
|
|
|
|
cdef size_t _make_nva(cnghttp2.nghttp2_nv **nva_ptr, headers):
|
|
cdef cnghttp2.nghttp2_nv *nva
|
|
cdef size_t nvlen
|
|
|
|
nvlen = len(headers)
|
|
nva = <cnghttp2.nghttp2_nv*>malloc(sizeof(cnghttp2.nghttp2_nv) * nvlen)
|
|
for i, (k, v) in enumerate(headers):
|
|
nva[i].name = k
|
|
nva[i].namelen = len(k)
|
|
nva[i].value = v
|
|
nva[i].valuelen = len(v)
|
|
nva[i].flags = cnghttp2.NGHTTP2_NV_FLAG_NONE
|
|
|
|
nva_ptr[0] = nva
|
|
|
|
return nvlen
|
|
|
|
cdef int server_on_header(cnghttp2.nghttp2_session *session,
|
|
const cnghttp2.nghttp2_frame *frame,
|
|
const uint8_t *name, size_t namelen,
|
|
const uint8_t *value, size_t valuelen,
|
|
uint8_t flags,
|
|
void *user_data):
|
|
cdef http2 = <_HTTP2SessionCoreBase>user_data
|
|
logging.debug('server_on_header, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)
|
|
|
|
handler = _get_stream_user_data(session, frame.hd.stream_id)
|
|
return on_header(name, namelen, value, valuelen, flags, handler)
|
|
|
|
cdef int client_on_header(cnghttp2.nghttp2_session *session,
|
|
const cnghttp2.nghttp2_frame *frame,
|
|
const uint8_t *name, size_t namelen,
|
|
const uint8_t *value, size_t valuelen,
|
|
uint8_t flags,
|
|
void *user_data):
|
|
cdef http2 = <_HTTP2SessionCoreBase>user_data
|
|
logging.debug('client_on_header, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)
|
|
|
|
if frame.hd.type == cnghttp2.NGHTTP2_HEADERS:
|
|
handler = _get_stream_user_data(session, frame.hd.stream_id)
|
|
elif frame.hd.type == cnghttp2.NGHTTP2_PUSH_PROMISE:
|
|
handler = _get_stream_user_data(session, frame.push_promise.promised_stream_id)
|
|
|
|
return on_header(name, namelen, value, valuelen, flags, handler)
|
|
|
|
|
|
cdef int on_header(const uint8_t *name, size_t namelen,
|
|
const uint8_t *value, size_t valuelen,
|
|
uint8_t flags,
|
|
object handler):
|
|
if not handler:
|
|
return 0
|
|
|
|
key = name[:namelen]
|
|
values = value[:valuelen].split(b'\x00')
|
|
if key == b':scheme':
|
|
handler.scheme = values[0]
|
|
elif key == b':method':
|
|
handler.method = values[0]
|
|
elif key == b':authority' or key == b'host':
|
|
handler.host = values[0]
|
|
elif key == b':path':
|
|
handler.path = values[0]
|
|
elif key == b':status':
|
|
handler.status = values[0]
|
|
|
|
if key == b'cookie':
|
|
handler.cookies.extend(values)
|
|
else:
|
|
for v in values:
|
|
handler.headers.append((key, v))
|
|
|
|
return 0
|
|
|
|
cdef int server_on_begin_request_headers(cnghttp2.nghttp2_session *session,
|
|
const cnghttp2.nghttp2_frame *frame,
|
|
void *user_data):
|
|
cdef http2 = <_HTTP2SessionCore>user_data
|
|
|
|
handler = http2._make_handler(frame.hd.stream_id)
|
|
cnghttp2.nghttp2_session_set_stream_user_data(session, frame.hd.stream_id,
|
|
<void*>handler)
|
|
|
|
return 0
|
|
|
|
cdef int server_on_begin_headers(cnghttp2.nghttp2_session *session,
|
|
const cnghttp2.nghttp2_frame *frame,
|
|
void *user_data):
|
|
if frame.hd.type == cnghttp2.NGHTTP2_HEADERS:
|
|
if frame.headers.cat == cnghttp2.NGHTTP2_HCAT_REQUEST:
|
|
return server_on_begin_request_headers(session, frame, user_data)
|
|
|
|
return 0
|
|
|
|
cdef int server_on_frame_recv(cnghttp2.nghttp2_session *session,
|
|
const cnghttp2.nghttp2_frame *frame,
|
|
void *user_data):
|
|
cdef http2 = <_HTTP2SessionCore>user_data
|
|
logging.debug('server_on_frame_recv, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)
|
|
|
|
if frame.hd.type == cnghttp2.NGHTTP2_DATA:
|
|
if frame.hd.flags & cnghttp2.NGHTTP2_FLAG_END_STREAM:
|
|
handler = _get_stream_user_data(session, frame.hd.stream_id)
|
|
if not handler:
|
|
return 0
|
|
try:
|
|
handler.on_request_done()
|
|
except:
|
|
sys.stderr.write(traceback.format_exc())
|
|
return http2._rst_stream(frame.hd.stream_id)
|
|
elif frame.hd.type == cnghttp2.NGHTTP2_HEADERS:
|
|
if frame.headers.cat == cnghttp2.NGHTTP2_HCAT_REQUEST:
|
|
handler = _get_stream_user_data(session, frame.hd.stream_id)
|
|
if not handler:
|
|
return 0
|
|
if handler.cookies:
|
|
handler.headers.append((b'cookie',
|
|
b'; '.join(handler.cookies)))
|
|
handler.cookies = None
|
|
try:
|
|
handler.on_headers()
|
|
if frame.hd.flags & cnghttp2.NGHTTP2_FLAG_END_STREAM:
|
|
handler.on_request_done()
|
|
except:
|
|
sys.stderr.write(traceback.format_exc())
|
|
return http2._rst_stream(frame.hd.stream_id)
|
|
elif frame.hd.type == cnghttp2.NGHTTP2_SETTINGS:
|
|
if (frame.hd.flags & cnghttp2.NGHTTP2_FLAG_ACK):
|
|
http2._stop_settings_timer()
|
|
|
|
return 0
|
|
|
|
cdef int on_data_chunk_recv(cnghttp2.nghttp2_session *session,
|
|
uint8_t flags,
|
|
int32_t stream_id, const uint8_t *data,
|
|
size_t length, void *user_data):
|
|
cdef http2 = <_HTTP2SessionCoreBase>user_data
|
|
|
|
handler = _get_stream_user_data(session, stream_id)
|
|
if not handler:
|
|
return 0
|
|
|
|
try:
|
|
handler.on_data(data[:length])
|
|
except:
|
|
sys.stderr.write(traceback.format_exc())
|
|
return http2._rst_stream(stream_id)
|
|
|
|
return 0
|
|
|
|
cdef int server_on_frame_send(cnghttp2.nghttp2_session *session,
|
|
const cnghttp2.nghttp2_frame *frame,
|
|
void *user_data):
|
|
cdef http2 = <_HTTP2SessionCore>user_data
|
|
logging.debug('server_on_frame_send, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)
|
|
|
|
if frame.hd.type == cnghttp2.NGHTTP2_PUSH_PROMISE:
|
|
# For PUSH_PROMISE, send push response immediately
|
|
handler = _get_stream_user_data\
|
|
(session, frame.push_promise.promised_stream_id)
|
|
if not handler:
|
|
return 0
|
|
|
|
http2.send_response(handler)
|
|
elif frame.hd.type == cnghttp2.NGHTTP2_SETTINGS:
|
|
if (frame.hd.flags & cnghttp2.NGHTTP2_FLAG_ACK) != 0:
|
|
return 0
|
|
http2._start_settings_timer()
|
|
elif frame.hd.type == cnghttp2.NGHTTP2_HEADERS:
|
|
if (frame.hd.flags & cnghttp2.NGHTTP2_FLAG_END_STREAM) and \
|
|
cnghttp2.nghttp2_session_check_server_session(session):
|
|
# Send RST_STREAM if remote is not closed yet
|
|
if cnghttp2.nghttp2_session_get_stream_remote_close(
|
|
session, frame.hd.stream_id) == 0:
|
|
http2._rst_stream(frame.hd.stream_id, cnghttp2.NGHTTP2_NO_ERROR)
|
|
|
|
cdef int server_on_frame_not_send(cnghttp2.nghttp2_session *session,
|
|
const cnghttp2.nghttp2_frame *frame,
|
|
int lib_error_code,
|
|
void *user_data):
|
|
cdef http2 = <_HTTP2SessionCore>user_data
|
|
logging.debug('server_on_frame_not_send, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)
|
|
|
|
if frame.hd.type == cnghttp2.NGHTTP2_PUSH_PROMISE:
|
|
# We have to remove handler here. Without this, it is not
|
|
# removed until session is terminated.
|
|
handler = _get_stream_user_data\
|
|
(session, frame.push_promise.promised_stream_id)
|
|
if not handler:
|
|
return 0
|
|
http2._remove_handler(handler)
|
|
|
|
cdef int on_stream_close(cnghttp2.nghttp2_session *session,
|
|
int32_t stream_id,
|
|
uint32_t error_code,
|
|
void *user_data):
|
|
cdef http2 = <_HTTP2SessionCoreBase>user_data
|
|
logging.debug('on_stream_close, stream_id:%s', stream_id)
|
|
|
|
handler = _get_stream_user_data(session, stream_id)
|
|
if not handler:
|
|
return 0
|
|
|
|
try:
|
|
handler.on_close(error_code)
|
|
except:
|
|
sys.stderr.write(traceback.format_exc())
|
|
|
|
http2._remove_handler(handler)
|
|
|
|
return 0
|
|
|
|
cdef ssize_t data_source_read(cnghttp2.nghttp2_session *session,
|
|
int32_t stream_id,
|
|
uint8_t *buf, size_t length,
|
|
uint32_t *data_flags,
|
|
cnghttp2.nghttp2_data_source *source,
|
|
void *user_data):
|
|
cdef http2 = <_HTTP2SessionCoreBase>user_data
|
|
generator = <object>source.ptr
|
|
|
|
http2.enter_callback()
|
|
try:
|
|
data, flag = generator(length)
|
|
except:
|
|
sys.stderr.write(traceback.format_exc())
|
|
return cnghttp2.NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE;
|
|
finally:
|
|
http2.leave_callback()
|
|
|
|
if flag == DATA_DEFERRED:
|
|
return cnghttp2.NGHTTP2_ERR_DEFERRED
|
|
|
|
if data:
|
|
nread = len(data)
|
|
memcpy(buf, <uint8_t*>data, nread)
|
|
else:
|
|
nread = 0
|
|
|
|
if flag == DATA_EOF:
|
|
data_flags[0] = cnghttp2.NGHTTP2_DATA_FLAG_EOF
|
|
if cnghttp2.nghttp2_session_check_server_session(session):
|
|
# Send RST_STREAM if remote is not closed yet
|
|
if cnghttp2.nghttp2_session_get_stream_remote_close(
|
|
session, stream_id) == 0:
|
|
http2._rst_stream(stream_id, cnghttp2.NGHTTP2_NO_ERROR)
|
|
elif flag != DATA_OK:
|
|
return cnghttp2.NGHTTP2_ERR_TEMPORAL_CALLBACK_FAILURE
|
|
|
|
return nread
|
|
|
|
cdef int client_on_begin_headers(cnghttp2.nghttp2_session *session,
|
|
const cnghttp2.nghttp2_frame *frame,
|
|
void *user_data):
|
|
cdef http2 = <_HTTP2ClientSessionCore>user_data
|
|
|
|
if frame.hd.type == cnghttp2.NGHTTP2_PUSH_PROMISE:
|
|
# Generate a temporary handler until the headers are all received
|
|
push_handler = BaseResponseHandler()
|
|
http2._add_handler(push_handler, frame.push_promise.promised_stream_id)
|
|
cnghttp2.nghttp2_session_set_stream_user_data(session, frame.push_promise.promised_stream_id,
|
|
<void*>push_handler)
|
|
|
|
return 0
|
|
|
|
cdef int client_on_frame_recv(cnghttp2.nghttp2_session *session,
|
|
const cnghttp2.nghttp2_frame *frame,
|
|
void *user_data):
|
|
cdef http2 = <_HTTP2ClientSessionCore>user_data
|
|
logging.debug('client_on_frame_recv, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)
|
|
|
|
if frame.hd.type == cnghttp2.NGHTTP2_DATA:
|
|
if frame.hd.flags & cnghttp2.NGHTTP2_FLAG_END_STREAM:
|
|
handler = _get_stream_user_data(session, frame.hd.stream_id)
|
|
if not handler:
|
|
return 0
|
|
try:
|
|
handler.on_response_done()
|
|
except:
|
|
sys.stderr.write(traceback.format_exc())
|
|
return http2._rst_stream(frame.hd.stream_id)
|
|
elif frame.hd.type == cnghttp2.NGHTTP2_HEADERS:
|
|
if frame.headers.cat == cnghttp2.NGHTTP2_HCAT_RESPONSE or frame.headers.cat == cnghttp2.NGHTTP2_HCAT_PUSH_RESPONSE:
|
|
handler = _get_stream_user_data(session, frame.hd.stream_id)
|
|
|
|
if not handler:
|
|
return 0
|
|
# TODO handle 1xx non-final response
|
|
if handler.cookies:
|
|
handler.headers.append((b'cookie',
|
|
b'; '.join(handler.cookies)))
|
|
handler.cookies = None
|
|
try:
|
|
handler.on_headers()
|
|
if frame.hd.flags & cnghttp2.NGHTTP2_FLAG_END_STREAM:
|
|
handler.on_response_done()
|
|
except:
|
|
sys.stderr.write(traceback.format_exc())
|
|
return http2._rst_stream(frame.hd.stream_id)
|
|
elif frame.hd.type == cnghttp2.NGHTTP2_SETTINGS:
|
|
if (frame.hd.flags & cnghttp2.NGHTTP2_FLAG_ACK):
|
|
http2._stop_settings_timer()
|
|
elif frame.hd.type == cnghttp2.NGHTTP2_PUSH_PROMISE:
|
|
handler = _get_stream_user_data(session, frame.hd.stream_id)
|
|
if not handler:
|
|
return 0
|
|
# Get the temporary push_handler which now should have all of the header data
|
|
push_handler = _get_stream_user_data(session, frame.push_promise.promised_stream_id)
|
|
if not push_handler:
|
|
return 0
|
|
# Remove the temporary handler
|
|
http2._remove_handler(push_handler)
|
|
cnghttp2.nghttp2_session_set_stream_user_data(session, frame.push_promise.promised_stream_id,
|
|
<void*>NULL)
|
|
|
|
try:
|
|
handler.on_push_promise(push_handler)
|
|
except:
|
|
sys.stderr.write(traceback.format_exc())
|
|
return http2._rst_stream(frame.hd.stream_id)
|
|
|
|
return 0
|
|
|
|
cdef int client_on_frame_send(cnghttp2.nghttp2_session *session,
|
|
const cnghttp2.nghttp2_frame *frame,
|
|
void *user_data):
|
|
cdef http2 = <_HTTP2ClientSessionCore>user_data
|
|
logging.debug('client_on_frame_send, type:%s, stream_id:%s', frame.hd.type, frame.hd.stream_id)
|
|
|
|
if frame.hd.type == cnghttp2.NGHTTP2_SETTINGS:
|
|
if (frame.hd.flags & cnghttp2.NGHTTP2_FLAG_ACK) != 0:
|
|
return 0
|
|
http2._start_settings_timer()
|
|
|
|
cdef class _HTTP2SessionCoreBase:
|
|
cdef cnghttp2.nghttp2_session *session
|
|
cdef transport
|
|
cdef handler_class
|
|
cdef handlers
|
|
cdef settings_timer
|
|
cdef inside_callback
|
|
|
|
def __cinit__(self, transport, handler_class=None):
|
|
self.session = NULL
|
|
self.transport = transport
|
|
self.handler_class = handler_class
|
|
self.handlers = set()
|
|
self.settings_timer = None
|
|
self.inside_callback = False
|
|
|
|
def __dealloc__(self):
|
|
cnghttp2.nghttp2_session_del(self.session)
|
|
|
|
def data_received(self, data):
|
|
cdef ssize_t rv
|
|
|
|
rv = cnghttp2.nghttp2_session_mem_recv(self.session, data, len(data))
|
|
if rv < 0:
|
|
raise Exception('nghttp2_session_mem_recv failed: {}'.format\
|
|
(_strerror(rv)))
|
|
self.send_data()
|
|
|
|
OUTBUF_MAX = 65535
|
|
SETTINGS_TIMEOUT = 5.0
|
|
|
|
def send_data(self):
|
|
cdef ssize_t outbuflen
|
|
cdef const uint8_t *outbuf
|
|
|
|
while True:
|
|
if self.transport.get_write_buffer_size() > self.OUTBUF_MAX:
|
|
break
|
|
outbuflen = cnghttp2.nghttp2_session_mem_send(self.session, &outbuf)
|
|
if outbuflen == 0:
|
|
break
|
|
if outbuflen < 0:
|
|
raise Exception('nghttp2_session_mem_send failed: {}'.format\
|
|
(_strerror(outbuflen)))
|
|
self.transport.write(outbuf[:outbuflen])
|
|
|
|
if self.transport.get_write_buffer_size() == 0 and \
|
|
cnghttp2.nghttp2_session_want_read(self.session) == 0 and \
|
|
cnghttp2.nghttp2_session_want_write(self.session) == 0:
|
|
self.transport.close()
|
|
|
|
def resume(self, stream_id):
|
|
cnghttp2.nghttp2_session_resume_data(self.session, stream_id)
|
|
if not self.inside_callback:
|
|
self.send_data()
|
|
|
|
def enter_callback(self):
|
|
self.inside_callback = True
|
|
|
|
def leave_callback(self):
|
|
self.inside_callback = False
|
|
|
|
def _make_handler(self, stream_id):
|
|
logging.debug('_make_handler, stream_id:%s', stream_id)
|
|
handler = self.handler_class(self, stream_id)
|
|
self.handlers.add(handler)
|
|
return handler
|
|
|
|
def _remove_handler(self, handler):
|
|
logging.debug('_remove_handler, stream_id:%s', handler.stream_id)
|
|
self.handlers.remove(handler)
|
|
|
|
def _add_handler(self, handler, stream_id):
|
|
logging.debug('_add_handler, stream_id:%s', stream_id)
|
|
handler.stream_id = stream_id
|
|
handler.http2 = self
|
|
handler.remote_address = self._get_remote_address()
|
|
handler.client_certificate = self._get_client_certificate()
|
|
self.handlers.add(handler)
|
|
|
|
def _rst_stream(self, stream_id,
|
|
error_code=cnghttp2.NGHTTP2_INTERNAL_ERROR):
|
|
cdef int rv
|
|
|
|
rv = cnghttp2.nghttp2_submit_rst_stream\
|
|
(self.session, cnghttp2.NGHTTP2_FLAG_NONE,
|
|
stream_id, error_code)
|
|
|
|
return rv
|
|
|
|
def _get_remote_address(self):
|
|
return self.transport.get_extra_info('peername')
|
|
|
|
def _get_client_certificate(self):
|
|
sock = self.transport.get_extra_info('socket')
|
|
try:
|
|
return sock.getpeercert()
|
|
except AttributeError:
|
|
return None
|
|
|
|
def _start_settings_timer(self):
|
|
loop = asyncio.get_event_loop()
|
|
self.settings_timer = loop.call_later(self.SETTINGS_TIMEOUT,
|
|
self._settings_timeout)
|
|
|
|
def _stop_settings_timer(self):
|
|
if self.settings_timer:
|
|
self.settings_timer.cancel()
|
|
self.settings_timer = None
|
|
|
|
def _settings_timeout(self):
|
|
cdef int rv
|
|
|
|
logging.debug('_settings_timeout')
|
|
|
|
self.settings_timer = None
|
|
|
|
rv = cnghttp2.nghttp2_session_terminate_session\
|
|
(self.session, cnghttp2.NGHTTP2_SETTINGS_TIMEOUT)
|
|
try:
|
|
self.send_data()
|
|
except Exception as err:
|
|
sys.stderr.write(traceback.format_exc())
|
|
self.transport.close()
|
|
return
|
|
|
|
def _log_request(self, handler):
|
|
now = datetime.datetime.now()
|
|
tv = time.mktime(now.timetuple())
|
|
datestr = email.utils.formatdate(timeval=tv, localtime=False,
|
|
usegmt=True)
|
|
try:
|
|
method = handler.method.decode('utf-8')
|
|
except:
|
|
method = handler.method
|
|
try:
|
|
path = handler.path.decode('utf-8')
|
|
except:
|
|
path = handler.path
|
|
logging.info('%s - - [%s] "%s %s HTTP/2" %s - %s', handler.remote_address[0],
|
|
datestr, method, path, handler.status,
|
|
'P' if handler.pushed else '-')
|
|
|
|
def close(self):
|
|
rv = cnghttp2.nghttp2_session_terminate_session\
|
|
(self.session, cnghttp2.NGHTTP2_NO_ERROR)
|
|
try:
|
|
self.send_data()
|
|
except Exception as err:
|
|
sys.stderr.write(traceback.format_exc())
|
|
self.transport.close()
|
|
return
|
|
|
|
cdef class _HTTP2SessionCore(_HTTP2SessionCoreBase):
|
|
def __cinit__(self, *args, **kwargs):
|
|
cdef cnghttp2.nghttp2_session_callbacks *callbacks
|
|
cdef cnghttp2.nghttp2_settings_entry iv[2]
|
|
cdef int rv
|
|
|
|
super(_HTTP2SessionCore, self).__init__(*args, **kwargs)
|
|
|
|
rv = cnghttp2.nghttp2_session_callbacks_new(&callbacks)
|
|
|
|
if rv != 0:
|
|
raise Exception('nghttp2_session_callbacks_new failed: {}'.format\
|
|
(_strerror(rv)))
|
|
|
|
cnghttp2.nghttp2_session_callbacks_set_on_header_callback(
|
|
callbacks, server_on_header)
|
|
cnghttp2.nghttp2_session_callbacks_set_on_begin_headers_callback(
|
|
callbacks, server_on_begin_headers)
|
|
cnghttp2.nghttp2_session_callbacks_set_on_frame_recv_callback(
|
|
callbacks, server_on_frame_recv)
|
|
cnghttp2.nghttp2_session_callbacks_set_on_stream_close_callback(
|
|
callbacks, on_stream_close)
|
|
cnghttp2.nghttp2_session_callbacks_set_on_frame_send_callback(
|
|
callbacks, server_on_frame_send)
|
|
cnghttp2.nghttp2_session_callbacks_set_on_frame_not_send_callback(
|
|
callbacks, server_on_frame_not_send)
|
|
cnghttp2.nghttp2_session_callbacks_set_on_data_chunk_recv_callback(
|
|
callbacks, on_data_chunk_recv)
|
|
|
|
rv = cnghttp2.nghttp2_session_server_new(&self.session, callbacks,
|
|
<void*>self)
|
|
|
|
cnghttp2.nghttp2_session_callbacks_del(callbacks)
|
|
|
|
if rv != 0:
|
|
raise Exception('nghttp2_session_server_new failed: {}'.format\
|
|
(_strerror(rv)))
|
|
|
|
iv[0].settings_id = cnghttp2.NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS
|
|
iv[0].value = 100
|
|
iv[1].settings_id = cnghttp2.NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE
|
|
iv[1].value = cnghttp2.NGHTTP2_INITIAL_WINDOW_SIZE
|
|
|
|
rv = cnghttp2.nghttp2_submit_settings(self.session,
|
|
cnghttp2.NGHTTP2_FLAG_NONE,
|
|
iv, sizeof(iv) // sizeof(iv[0]))
|
|
|
|
if rv != 0:
|
|
raise Exception('nghttp2_submit_settings failed: {}'.format\
|
|
(_strerror(rv)))
|
|
|
|
def send_response(self, handler):
|
|
cdef cnghttp2.nghttp2_data_provider prd
|
|
cdef cnghttp2.nghttp2_data_provider *prd_ptr
|
|
cdef cnghttp2.nghttp2_nv *nva
|
|
cdef size_t nvlen
|
|
cdef int rv
|
|
|
|
logging.debug('send_response, stream_id:%s', handler.stream_id)
|
|
|
|
nva = NULL
|
|
nvlen = _make_nva(&nva, handler.response_headers)
|
|
|
|
if handler.response_body:
|
|
prd.source.ptr = <void*>handler.response_body
|
|
prd.read_callback = data_source_read
|
|
prd_ptr = &prd
|
|
else:
|
|
prd_ptr = NULL
|
|
|
|
rv = cnghttp2.nghttp2_submit_response(self.session, handler.stream_id,
|
|
nva, nvlen, prd_ptr)
|
|
|
|
free(nva)
|
|
|
|
if rv != 0:
|
|
# TODO Ignore return value
|
|
self._rst_stream(handler.stream_id)
|
|
raise Exception('nghttp2_submit_response failed: {}'.format\
|
|
(_strerror(rv)))
|
|
|
|
self._log_request(handler)
|
|
|
|
def push(self, handler, promised_handler):
|
|
cdef cnghttp2.nghttp2_nv *nva
|
|
cdef size_t nvlen
|
|
cdef int32_t promised_stream_id
|
|
|
|
self.handlers.add(promised_handler)
|
|
|
|
nva = NULL
|
|
nvlen = _make_nva(&nva, promised_handler.headers)
|
|
|
|
promised_stream_id = cnghttp2.nghttp2_submit_push_promise\
|
|
(self.session,
|
|
cnghttp2.NGHTTP2_FLAG_NONE,
|
|
handler.stream_id,
|
|
nva, nvlen,
|
|
<void*>promised_handler)
|
|
if promised_stream_id < 0:
|
|
raise Exception('nghttp2_submit_push_promise failed: {}'.format\
|
|
(_strerror(promised_stream_id)))
|
|
|
|
promised_handler.stream_id = promised_stream_id
|
|
|
|
logging.debug('push, stream_id:%s', promised_stream_id)
|
|
|
|
return promised_handler
|
|
|
|
def connection_lost(self):
|
|
self._stop_settings_timer()
|
|
|
|
for handler in self.handlers:
|
|
handler.on_close(cnghttp2.NGHTTP2_INTERNAL_ERROR)
|
|
self.handlers = set()
|
|
|
|
cdef class _HTTP2ClientSessionCore(_HTTP2SessionCoreBase):
|
|
def __cinit__(self, *args, **kwargs):
|
|
cdef cnghttp2.nghttp2_session_callbacks *callbacks
|
|
cdef cnghttp2.nghttp2_settings_entry iv[2]
|
|
cdef int rv
|
|
|
|
super(_HTTP2ClientSessionCore, self).__init__(*args, **kwargs)
|
|
|
|
rv = cnghttp2.nghttp2_session_callbacks_new(&callbacks)
|
|
|
|
if rv != 0:
|
|
raise Exception('nghttp2_session_callbacks_new failed: {}'.format\
|
|
(_strerror(rv)))
|
|
|
|
cnghttp2.nghttp2_session_callbacks_set_on_header_callback(
|
|
callbacks, client_on_header)
|
|
cnghttp2.nghttp2_session_callbacks_set_on_begin_headers_callback(
|
|
callbacks, client_on_begin_headers)
|
|
cnghttp2.nghttp2_session_callbacks_set_on_frame_recv_callback(
|
|
callbacks, client_on_frame_recv)
|
|
cnghttp2.nghttp2_session_callbacks_set_on_stream_close_callback(
|
|
callbacks, on_stream_close)
|
|
cnghttp2.nghttp2_session_callbacks_set_on_frame_send_callback(
|
|
callbacks, client_on_frame_send)
|
|
cnghttp2.nghttp2_session_callbacks_set_on_data_chunk_recv_callback(
|
|
callbacks, on_data_chunk_recv)
|
|
|
|
rv = cnghttp2.nghttp2_session_client_new(&self.session, callbacks,
|
|
<void*>self)
|
|
|
|
cnghttp2.nghttp2_session_callbacks_del(callbacks)
|
|
|
|
if rv != 0:
|
|
raise Exception('nghttp2_session_client_new failed: {}'.format\
|
|
(_strerror(rv)))
|
|
|
|
iv[0].settings_id = cnghttp2.NGHTTP2_SETTINGS_MAX_CONCURRENT_STREAMS
|
|
iv[0].value = 100
|
|
iv[1].settings_id = cnghttp2.NGHTTP2_SETTINGS_INITIAL_WINDOW_SIZE
|
|
iv[1].value = cnghttp2.NGHTTP2_INITIAL_WINDOW_SIZE
|
|
|
|
rv = cnghttp2.nghttp2_submit_settings(self.session,
|
|
cnghttp2.NGHTTP2_FLAG_NONE,
|
|
iv, sizeof(iv) // sizeof(iv[0]))
|
|
|
|
if rv != 0:
|
|
raise Exception('nghttp2_submit_settings failed: {}'.format\
|
|
(_strerror(rv)))
|
|
|
|
def send_request(self, method, scheme, host, path, headers, body, handler):
|
|
cdef cnghttp2.nghttp2_data_provider prd
|
|
cdef cnghttp2.nghttp2_data_provider *prd_ptr
|
|
cdef cnghttp2.nghttp2_priority_spec *pri_ptr
|
|
cdef cnghttp2.nghttp2_nv *nva
|
|
cdef size_t nvlen
|
|
cdef int32_t stream_id
|
|
|
|
body = wrap_body(body)
|
|
|
|
custom_headers = _encode_headers(headers)
|
|
headers = [
|
|
(b':method', method.encode('utf-8')),
|
|
(b':scheme', scheme.encode('utf-8')),
|
|
(b':authority', host.encode('utf-8')),
|
|
(b':path', path.encode('utf-8'))
|
|
]
|
|
headers.extend(custom_headers)
|
|
|
|
nva = NULL
|
|
nvlen = _make_nva(&nva, headers)
|
|
|
|
if body:
|
|
prd.source.ptr = <void*>body
|
|
prd.read_callback = data_source_read
|
|
prd_ptr = &prd
|
|
else:
|
|
prd_ptr = NULL
|
|
|
|
# TODO: Enable priorities
|
|
pri_ptr = NULL
|
|
|
|
stream_id = cnghttp2.nghttp2_submit_request\
|
|
(self.session, pri_ptr,
|
|
nva, nvlen, prd_ptr,
|
|
<void*>handler)
|
|
free(nva)
|
|
|
|
if stream_id < 0:
|
|
raise Exception('nghttp2_submit_request failed: {}'.format\
|
|
(_strerror(stream_id)))
|
|
|
|
logging.debug('request, stream_id:%s', stream_id)
|
|
|
|
self._add_handler(handler, stream_id)
|
|
cnghttp2.nghttp2_session_set_stream_user_data(self.session, stream_id,
|
|
<void*>handler)
|
|
|
|
return handler
|
|
|
|
def push(self, push_promise, handler):
|
|
if handler:
|
|
# push_promise accepted, fill in the handler with the stored
|
|
# headers from the push_promise
|
|
handler.status = push_promise.status
|
|
handler.scheme = push_promise.scheme
|
|
handler.method = push_promise.method
|
|
handler.host = push_promise.host
|
|
handler.path = push_promise.path
|
|
handler.cookies = push_promise.cookies
|
|
handler.stream_id = push_promise.stream_id
|
|
handler.http2 = self
|
|
handler.pushed = True
|
|
|
|
self._add_handler(handler, handler.stream_id)
|
|
|
|
cnghttp2.nghttp2_session_set_stream_user_data(self.session, handler.stream_id,
|
|
<void*>handler)
|
|
else:
|
|
# push_promise rejected, reset the stream
|
|
self._rst_stream(push_promise.stream_id,
|
|
error_code=cnghttp2.NGHTTP2_NO_ERROR)
|
|
|
|
if asyncio:
|
|
|
|
class BaseRequestHandler:
|
|
|
|
"""HTTP/2 request (stream) handler base class.
|
|
|
|
The class is used to handle the HTTP/2 stream. By default, it does
|
|
nothing. It must be subclassed to handle each event callback method.
|
|
|
|
The first callback method invoked is on_headers(). It is called
|
|
when HEADERS frame, which includes request header fields, is
|
|
arrived.
|
|
|
|
If request has request body, on_data(data) is invoked for each
|
|
chunk of received data.
|
|
|
|
When whole request is received, on_request_done() is invoked.
|
|
|
|
When stream is closed, on_close(error_code) is called.
|
|
|
|
The application can send response using send_response() method. It
|
|
can be used in on_headers(), on_data() or on_request_done().
|
|
|
|
The application can push resource using push() method. It must be
|
|
used before send_response() call.
|
|
|
|
The following instance variables are available:
|
|
|
|
client_address
|
|
Contains a tuple of the form (host, port) referring to the client's
|
|
address.
|
|
|
|
client_certificate
|
|
May contain the client certificate in its non-binary form
|
|
|
|
stream_id
|
|
Stream ID of this stream
|
|
|
|
scheme
|
|
Scheme of the request URI. This is a value of :scheme header field.
|
|
|
|
method
|
|
Method of this stream. This is a value of :method header field.
|
|
|
|
host
|
|
This is a value of :authority or host header field.
|
|
|
|
path
|
|
This is a value of :path header field.
|
|
|
|
headers
|
|
Request header fields
|
|
|
|
"""
|
|
|
|
def __init__(self, http2, stream_id):
|
|
self.headers = []
|
|
self.cookies = []
|
|
# Stream ID. For promised stream, it is initially -1.
|
|
self.stream_id = stream_id
|
|
self.http2 = http2
|
|
# address of the client
|
|
self.remote_address = self.http2._get_remote_address()
|
|
# certificate of the client
|
|
self._client_certificate = self.http2._get_client_certificate()
|
|
# :scheme header field in request
|
|
self.scheme = None
|
|
# :method header field in request
|
|
self.method = None
|
|
# :authority or host header field in request
|
|
self.host = None
|
|
# :path header field in request
|
|
self.path = None
|
|
# HTTP status
|
|
self.status = None
|
|
# True if this is a handler for pushed resource
|
|
self.pushed = False
|
|
|
|
@property
|
|
def client_address(self):
|
|
return self.remote_address
|
|
|
|
@property
|
|
def client_certificate(self):
|
|
return self._client_certificate
|
|
|
|
def on_headers(self):
|
|
|
|
'''Called when request HEADERS is arrived.
|
|
|
|
'''
|
|
pass
|
|
|
|
def on_data(self, data):
|
|
|
|
'''Called when a chunk of request body is arrived. This method
|
|
will be called multiple times until all data are received.
|
|
|
|
'''
|
|
pass
|
|
|
|
def on_request_done(self):
|
|
|
|
'''Called when whole request was received
|
|
|
|
'''
|
|
pass
|
|
|
|
def on_close(self, error_code):
|
|
|
|
'''Called when stream is about to close.
|
|
|
|
'''
|
|
pass
|
|
|
|
def send_response(self, status=200, headers=None, body=None):
|
|
|
|
'''Send response. The status is HTTP status code. The headers is
|
|
additional response headers. The :status header field is
|
|
appended by the library. The body is the response body. It
|
|
could be None if response body is empty. Or it must be
|
|
instance of either str, bytes, io.IOBase or callable,
|
|
called body generator, which takes one parameter,
|
|
size. The body generator generates response body. It can
|
|
pause generation of response so that it can wait for slow
|
|
backend data generation. When invoked, it should return
|
|
tuple, byte string and flag. The flag is either DATA_OK,
|
|
DATA_EOF and DATA_DEFERRED. For non-empty byte string and
|
|
it is not the last chunk of response, DATA_OK is returned
|
|
as flag. If this is the last chunk of the response (byte
|
|
string is possibly None), DATA_EOF must be returned as
|
|
flag. If there is no data available right now, but
|
|
additional data are anticipated, return tuple (None,
|
|
DATA_DEFERRD). When data arrived, call resume() and
|
|
restart response body transmission.
|
|
|
|
Only the body generator can pause response body
|
|
generation; instance of io.IOBase must not block.
|
|
|
|
If instance of str is specified as body, it is encoded
|
|
using UTF-8.
|
|
|
|
The headers is a list of tuple of the form (name,
|
|
value). The name and value can be either unicode string or
|
|
byte string.
|
|
|
|
On error, exception will be thrown.
|
|
|
|
'''
|
|
if self.status is not None:
|
|
raise Exception('response has already been sent')
|
|
|
|
if not status:
|
|
raise Exception('status must not be empty')
|
|
|
|
body = wrap_body(body)
|
|
|
|
self._set_response_prop(status, headers, body)
|
|
self.http2.send_response(self)
|
|
|
|
def push(self, path, method='GET', request_headers=None,
|
|
status=200, headers=None, body=None):
|
|
|
|
'''Push a resource. The path is a path portion of request URI
|
|
for this
|
|
resource. The method is a method to access this
|
|
resource. The request_headers is additional request
|
|
headers to access this resource. The :scheme, :method,
|
|
:authority and :path are appended by the library. The
|
|
:scheme and :authority are inherited from the request (not
|
|
request_headers parameter).
|
|
|
|
The status is HTTP status code. The headers is additional
|
|
response headers. The :status header field is appended by
|
|
the library. The body is the response body. It has the
|
|
same semantics of body parameter of send_response().
|
|
|
|
The headers and request_headers are a list of tuple of the
|
|
form (name, value). The name and value can be either
|
|
unicode string or byte string.
|
|
|
|
On error, exception will be thrown.
|
|
|
|
'''
|
|
if not status:
|
|
raise Exception('status must not be empty')
|
|
|
|
if not method:
|
|
raise Exception('method must not be empty')
|
|
|
|
if not path:
|
|
raise Exception('path must not be empty')
|
|
|
|
body = wrap_body(body)
|
|
|
|
promised_handler = self.http2._make_handler(-1)
|
|
promised_handler.pushed = True
|
|
promised_handler.scheme = self.scheme
|
|
promised_handler.method = method.encode('utf-8')
|
|
promised_handler.host = self.host
|
|
promised_handler.path = path.encode('utf-8')
|
|
promised_handler._set_response_prop(status, headers, body)
|
|
|
|
headers = [
|
|
(b':method', promised_handler.method),
|
|
(b':scheme', promised_handler.scheme),
|
|
(b':authority', promised_handler.host),
|
|
(b':path', promised_handler.path)
|
|
]
|
|
headers.extend(_encode_headers(request_headers))
|
|
|
|
promised_handler.headers = headers
|
|
|
|
return self.http2.push(self, promised_handler)
|
|
|
|
def _set_response_prop(self, status, headers, body):
|
|
self.status = status
|
|
|
|
if headers is None:
|
|
headers = []
|
|
|
|
self.response_headers = [(b':status', str(status).encode('utf-8'))]
|
|
self.response_headers.extend(_encode_headers(headers))
|
|
|
|
self.response_body = body
|
|
|
|
def resume(self):
|
|
self.http2.resume(self.stream_id)
|
|
|
|
def _encode_headers(headers):
|
|
if not headers:
|
|
return []
|
|
return [(k if isinstance(k, bytes) else k.encode('utf-8'),
|
|
v if isinstance(v, bytes) else v.encode('utf-8')) \
|
|
for k, v in headers]
|
|
|
|
class _HTTP2Session(asyncio.Protocol):
|
|
|
|
def __init__(self, RequestHandlerClass):
|
|
asyncio.Protocol.__init__(self)
|
|
self.RequestHandlerClass = RequestHandlerClass
|
|
self.http2 = None
|
|
|
|
def connection_made(self, transport):
|
|
address = transport.get_extra_info('peername')
|
|
logging.info('connection_made, address:%s, port:%s', address[0], address[1])
|
|
|
|
self.transport = transport
|
|
sock = self.transport.get_extra_info('socket')
|
|
try:
|
|
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
|
|
except OSError as e:
|
|
logging.info('failed to set tcp-nodelay: %s', str(e))
|
|
ssl_ctx = self.transport.get_extra_info('sslcontext')
|
|
if ssl_ctx:
|
|
ssl_obj = self.transport.get_extra_info('ssl_object')
|
|
protocol = negotiated_protocol(ssl_obj)
|
|
if protocol is None or protocol.encode('utf-8') != \
|
|
cnghttp2.NGHTTP2_PROTO_VERSION_ID:
|
|
self.transport.abort()
|
|
return
|
|
try:
|
|
self.http2 = _HTTP2SessionCore\
|
|
(self.transport,
|
|
self.RequestHandlerClass)
|
|
except Exception as err:
|
|
sys.stderr.write(traceback.format_exc())
|
|
self.transport.abort()
|
|
return
|
|
|
|
|
|
def connection_lost(self, exc):
|
|
logging.info('connection_lost')
|
|
if self.http2:
|
|
self.http2.connection_lost()
|
|
self.http2 = None
|
|
|
|
def data_received(self, data):
|
|
try:
|
|
self.http2.data_received(data)
|
|
except Exception as err:
|
|
sys.stderr.write(traceback.format_exc())
|
|
self.transport.close()
|
|
return
|
|
|
|
def resume_writing(self):
|
|
try:
|
|
self.http2.send_data()
|
|
except Exception as err:
|
|
sys.stderr.write(traceback.format_exc())
|
|
self.transport.close()
|
|
return
|
|
|
|
class HTTP2Server:
|
|
|
|
'''HTTP/2 server.
|
|
|
|
This class builds on top of the asyncio event loop. On
|
|
construction, RequestHandlerClass must be given, which must be a
|
|
subclass of BaseRequestHandler class.
|
|
|
|
'''
|
|
def __init__(self, address, RequestHandlerClass, ssl=None):
|
|
|
|
'''address is a tuple of the listening address and port (e.g.,
|
|
('127.0.0.1', 8080)). RequestHandlerClass must be a subclass
|
|
of BaseRequestHandler class to handle a HTTP/2 stream. The
|
|
ssl can be ssl.SSLContext instance. If it is not None, the
|
|
resulting server is SSL/TLS capable.
|
|
|
|
'''
|
|
def session_factory():
|
|
return _HTTP2Session(RequestHandlerClass)
|
|
|
|
self.loop = asyncio.get_event_loop()
|
|
|
|
if ssl:
|
|
set_application_protocol(ssl)
|
|
|
|
coro = self.loop.create_server(session_factory,
|
|
host=address[0], port=address[1],
|
|
ssl=ssl)
|
|
self.server = self.loop.run_until_complete(coro)
|
|
logging.info('listen, address:%s, port:%s', address[0], address[1])
|
|
|
|
def serve_forever(self):
|
|
try:
|
|
self.loop.run_forever()
|
|
finally:
|
|
self.server.close()
|
|
self.loop.close()
|
|
|
|
|
|
|
|
class BaseResponseHandler:
|
|
|
|
"""HTTP/2 response (stream) handler base class.
|
|
|
|
The class is used to handle the HTTP/2 stream. By default, it does
|
|
not nothing. It must be subclassed to handle each event callback
|
|
method.
|
|
|
|
The first callback method invoked is on_headers(). It is called
|
|
when HEADERS frame, which includes response header fields, is
|
|
arrived.
|
|
|
|
If response has a body, on_data(data) is invoked for each
|
|
chunk of received data.
|
|
|
|
When whole response is received, on_response_done() is invoked.
|
|
|
|
When stream is closed or underlying connection is lost,
|
|
on_close(error_code) is called.
|
|
|
|
The application can send follow up requests using HTTP2Client.send_request() method.
|
|
|
|
The application can handle push resource using on_push_promise() method.
|
|
|
|
The following instance variables are available:
|
|
|
|
server_address
|
|
Contains a tuple of the form (host, port) referring to the server's
|
|
address.
|
|
|
|
stream_id
|
|
Stream ID of this stream
|
|
|
|
scheme
|
|
Scheme of the request URI. This is a value of :scheme header field.
|
|
|
|
method
|
|
Method of this stream. This is a value of :method header field.
|
|
|
|
host
|
|
This is a value of :authority or host header field.
|
|
|
|
path
|
|
This is a value of :path header field.
|
|
|
|
headers
|
|
Response header fields. There is a special exception. If this
|
|
object is passed to push_promise(), this instance variable contains
|
|
pushed request header fields.
|
|
|
|
"""
|
|
|
|
def __init__(self, http2=None, stream_id=-1):
|
|
self.headers = []
|
|
self.cookies = []
|
|
# Stream ID. For promised stream, it is initially -1.
|
|
self.stream_id = stream_id
|
|
self.http2 = http2
|
|
# address of the server
|
|
self.remote_address = None
|
|
# :scheme header field in request
|
|
self.scheme = None
|
|
# :method header field in request
|
|
self.method = None
|
|
# :authority or host header field in request
|
|
self.host = None
|
|
# :path header field in request
|
|
self.path = None
|
|
# HTTP status
|
|
self.status = None
|
|
# True if this is a handler for pushed resource
|
|
self.pushed = False
|
|
|
|
@property
|
|
def server_address(self):
|
|
return self.remote_address
|
|
|
|
def on_headers(self):
|
|
|
|
'''Called when response HEADERS is arrived.
|
|
|
|
'''
|
|
pass
|
|
|
|
def on_data(self, data):
|
|
|
|
'''Called when a chunk of response body is arrived. This method
|
|
will be called multiple times until all data are received.
|
|
|
|
'''
|
|
pass
|
|
|
|
def on_response_done(self):
|
|
|
|
'''Called when whole response was received
|
|
|
|
'''
|
|
pass
|
|
|
|
def on_close(self, error_code):
|
|
|
|
'''Called when stream is about to close.
|
|
|
|
'''
|
|
pass
|
|
|
|
def on_push_promise(self, push_promise):
|
|
|
|
'''Called when a push is promised. Default behavior is to
|
|
cancel the push. If application overrides this method,
|
|
it should call either accept_push or reject_push.
|
|
|
|
'''
|
|
self.reject_push(push_promise)
|
|
|
|
def reject_push(self, push_promise):
|
|
|
|
'''Convenience method equivalent to calling accept_push
|
|
with a falsy value.
|
|
|
|
'''
|
|
self.http2.push(push_promise, None)
|
|
|
|
def accept_push(self, push_promise, handler=None):
|
|
|
|
'''Accept a push_promise and provider a handler for the
|
|
new stream. If a falsy value is supplied for the handler,
|
|
the push is rejected.
|
|
|
|
'''
|
|
self.http2.push(push_promise, handler)
|
|
|
|
def resume(self):
|
|
self.http2.resume(self.stream_id)
|
|
|
|
class _HTTP2ClientSession(asyncio.Protocol):
|
|
|
|
def __init__(self, client):
|
|
asyncio.Protocol.__init__(self)
|
|
self.http2 = None
|
|
self.pending = []
|
|
self.client = client
|
|
|
|
def connection_made(self, transport):
|
|
address = transport.get_extra_info('peername')
|
|
logging.info('connection_made, address:%s, port:%s', address[0], address[1])
|
|
|
|
self.transport = transport
|
|
sock = self.transport.get_extra_info('socket')
|
|
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
|
|
ssl_ctx = self.transport.get_extra_info('sslcontext')
|
|
if ssl_ctx:
|
|
ssl_obj = self.transport.get_extra_info('ssl_object')
|
|
protocol = negotiated_protocol(ssl_obj)
|
|
if protocol is None or protocol.encode('utf-8') != \
|
|
cnghttp2.NGHTTP2_PROTO_VERSION_ID:
|
|
self.transport.abort()
|
|
|
|
self.http2 = _HTTP2ClientSessionCore(self.transport)
|
|
|
|
# Clear pending requests
|
|
send_pending = self.pending
|
|
self.pending = []
|
|
for method,scheme,host,path,headers,body,handler in send_pending:
|
|
self.send_request(method=method, scheme=scheme, host=host, path=path,\
|
|
headers=headers, body=body, handler=handler)
|
|
self.http2.send_data()
|
|
|
|
def connection_lost(self, exc):
|
|
logging.info('connection_lost')
|
|
if self.http2:
|
|
self.http2 = None
|
|
self.client.close()
|
|
|
|
def data_received(self, data):
|
|
try:
|
|
self.http2.data_received(data)
|
|
except Exception as err:
|
|
sys.stderr.write(traceback.format_exc())
|
|
self.transport.close()
|
|
return
|
|
|
|
def resume_writing(self):
|
|
try:
|
|
self.http2.send_data()
|
|
except Exception as err:
|
|
sys.stderr.write(traceback.format_exc())
|
|
self.transport.close()
|
|
return
|
|
|
|
def send_request(self, method, scheme, host, path, headers, body, handler):
|
|
try:
|
|
# Waiting until connection established
|
|
if not self.http2:
|
|
self.pending.append([method, scheme, host, path, headers, body, handler])
|
|
return
|
|
|
|
self.http2.send_request(method=method, scheme=scheme, host=host, path=path,\
|
|
headers=headers, body=body, handler=handler)
|
|
self.http2.send_data()
|
|
except Exception as err:
|
|
sys.stderr.write(traceback.format_exc())
|
|
self.transport.close()
|
|
return
|
|
|
|
def close(self):
|
|
if self.http2:
|
|
self.http2.close()
|
|
|
|
|
|
class HTTP2Client:
|
|
|
|
'''HTTP/2 client.
|
|
|
|
This class builds on top of the asyncio event loop.
|
|
|
|
'''
|
|
def __init__(self, address, loop=None, ssl=None):
|
|
|
|
'''address is a tuple of the connect address and port (e.g.,
|
|
('127.0.0.1', 8080)). The ssl can be ssl.SSLContext instance.
|
|
If it is not None, the resulting client is SSL/TLS capable.
|
|
'''
|
|
|
|
self.address = address
|
|
self.session = _HTTP2ClientSession(self)
|
|
def session_factory():
|
|
return self.session
|
|
|
|
if ssl:
|
|
set_application_protocol(ssl)
|
|
|
|
self.loop = loop
|
|
if not self.loop:
|
|
self.loop = asyncio.get_event_loop()
|
|
|
|
coro = self.loop.create_connection(session_factory,
|
|
host=address[0], port=address[1],
|
|
ssl=ssl)
|
|
|
|
if ssl:
|
|
self.scheme = 'https'
|
|
else:
|
|
self.scheme = 'http'
|
|
|
|
self.transport,_ = self.loop.run_until_complete(coro)
|
|
logging.info('connect, address:%s, port:%s', self.address[0], self.address[1])
|
|
|
|
@property
|
|
def io_loop(self):
|
|
return self.loop
|
|
|
|
def close(self):
|
|
self.session.close()
|
|
|
|
def send_request(self, method='GET', url='/', headers=None, body=None, handler=None):
|
|
url = urlparse(url)
|
|
scheme = url.scheme if url.scheme else self.scheme
|
|
host = url.netloc if url.netloc else self.address[0]+':'+str(self.address[1])
|
|
path = url.path
|
|
if url.params:
|
|
path += ';'+url.params
|
|
if url.query:
|
|
path += '?'+url.query
|
|
if url.fragment:
|
|
path += '#'+url.fragment
|
|
|
|
self.session.send_request(method=method, scheme=scheme, host=host, path=path,\
|
|
headers=headers, body=body, handler=handler)
|