h2 server

h2_server.py

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

A fully-functional HTTP/2 server using asyncio. Requires Python 3.5+.

This example demonstrates handling requests with bodies, as well as handling
those without. In particular, it demonstrates the fact that DataReceived may
be called multiple times, and that applications must handle that possibility.
"""
import asyncio
import io
import json
import ssl
import collections
from typing import List, Tuple

from h2.config import H2Configuration
from h2.connection import H2Connection
from h2.events import (
    ConnectionTerminated, DataReceived, RemoteSettingsChanged,
    RequestReceived, StreamEnded, StreamReset, WindowUpdated
)
from h2.errors import ErrorCodes
from h2.exceptions import ProtocolError, StreamClosedError
from h2.settings import SettingCodes


RequestData = collections.namedtuple('RequestData', ['headers', 'data'])


class H2Protocol(asyncio.Protocol):
    def __init__(self):
        config = H2Configuration(client_side=False, header_encoding='utf-8')
        self.conn = H2Connection(config=config)
        self.transport = None
        self.stream_data = {}
        self.flow_control_futures = {}

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

    def connection_lost(self, exc):
        for future in self.flow_control_futures.values():
            future.cancel()
        self.flow_control_futures = {}

    def data_received(self, data: bytes):
        try:
            events = self.conn.receive_data(data)
        except ProtocolError as e:
            self.transport.write(self.conn.data_to_send())
            self.transport.close()
        else:
            self.transport.write(self.conn.data_to_send())
            for event in events:
                if isinstance(event, RequestReceived):
                    self.request_received(event.headers, event.stream_id)
                elif isinstance(event, DataReceived):
                    self.receive_data(event.data, event.stream_id)
                elif isinstance(event, StreamEnded):
                    self.stream_complete(event.stream_id)
                elif isinstance(event, ConnectionTerminated):
                    self.transport.close()
                elif isinstance(event, StreamReset):
                    self.stream_reset(event.stream_id)
                elif isinstance(event, WindowUpdated):
                    self.window_updated(event.stream_id, event.delta)
                elif isinstance(event, RemoteSettingsChanged):
                    if SettingCodes.INITIAL_WINDOW_SIZE in event.changed_settings:
                        self.window_updated(None, 0)

                self.transport.write(self.conn.data_to_send())

    def request_received(self, headers: List[Tuple[str, str]], stream_id: int):
        headers = collections.OrderedDict(headers)
        method = headers[':method']

        # Store off the request data.
        request_data = RequestData(headers, io.BytesIO())
        self.stream_data[stream_id] = request_data

    def stream_complete(self, stream_id: int):
        """
        When a stream is complete, we can send our response.
        """
        try:
            request_data = self.stream_data[stream_id]
        except KeyError:
            # Just return, we probably 405'd this already
            return

        headers = request_data.headers
        body = request_data.data.getvalue().decode('utf-8')

        data = json.dumps(
            {"headers": headers, "body": body}, indent=4
        ).encode("utf8")

        response_headers = (
            (':status', '200'),
            ('content-type', 'application/json'),
            ('content-length', str(len(data))),
            ('server', 'asyncio-h2'),
        )
        self.conn.send_headers(stream_id, response_headers)
        asyncio.ensure_future(self.send_data(data, stream_id))

    def receive_data(self, data: bytes, stream_id: int):
        """
        We've received some data on a stream. If that stream is one we're
        expecting data on, save it off. Otherwise, reset the stream.
        """
        try:
            stream_data = self.stream_data[stream_id]
        except KeyError:
            self.conn.reset_stream(
                stream_id, error_code=ErrorCodes.PROTOCOL_ERROR
            )
        else:
            stream_data.data.write(data)

    def stream_reset(self, stream_id):
        """
        A stream reset was sent. Stop sending data.
        """
        if stream_id in self.flow_control_futures:
            future = self.flow_control_futures.pop(stream_id)
            future.cancel()

    async def send_data(self, data, stream_id):
        """
        Send data according to the flow control rules.
        """
        while data:
            while self.conn.local_flow_control_window(stream_id) < 1:
                try:
                    await self.wait_for_flow_control(stream_id)
                except asyncio.CancelledError:
                    return

            chunk_size = min(
                self.conn.local_flow_control_window(stream_id),
                len(data),
                self.conn.max_outbound_frame_size,
            )

            try:
                self.conn.send_data(
                    stream_id,
                    data[:chunk_size],
                    end_stream=(chunk_size == len(data))
                )
            except (StreamClosedError, ProtocolError):
                # The stream got closed and we didn't get told. We're done
                # here.
                break

            self.transport.write(self.conn.data_to_send())
            data = data[chunk_size:]

    async def wait_for_flow_control(self, stream_id):
        """
        Waits for a Future that fires when the flow control window is opened.
        """
        f = asyncio.Future()
        self.flow_control_futures[stream_id] = f
        await f

    def window_updated(self, stream_id, delta):
        """
        A window update frame was received. Unblock some number of flow control
        Futures.
        """
        if stream_id and stream_id in self.flow_control_futures:
            f = self.flow_control_futures.pop(stream_id)
            f.set_result(delta)
        elif not stream_id:
            for f in self.flow_control_futures.values():
                f.set_result(delta)

            self.flow_control_futures = {}


ssl_context = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
ssl_context.options |= (
    ssl.OP_NO_COMPRESSION | ssl.OP_ENABLE_KTLS | ssl.OP_CIPHER_SERVER_PREFERENCE
    #ssl.OP_NO_TLSv1 | ssl.OP_NO_TLSv1_1 | ssl.OP_NO_TLSv1_3 | ssl.OP_NO_COMPRESSION
)
ssl_context.load_cert_chain(certfile="/home/kingluo/tempesta/server.crt", keyfile="/home/kingluo/tempesta/server.key")
ssl_context.set_alpn_protocols(["h2"])
ssl_context.set_ciphers('ECDHE-RSA-AES128-GCM-SHA256:ECDHE-ECDSA-AES128-GCM-SHA256')


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

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

# Close the server
server.close()
loop.run_until_complete(server.wait_closed())
loop.close()

main.go

package main

import (
	"encoding/binary"
	"flag"
	"log"
	"crypto/tls"
	"time"
	"net"

	"golang.org/x/net/http2"
)

const (
	transportDefaultStreamFlow = 4 << 20
	maxFrameSize               = 1<<24 - 1
	maxHeaderListSize          = 10 << 20
	headerTableSize            = 4096
)

var (
	connections     int
	threads    	int
	address         string
	host       	string
	debug		int
)

func init() {
	flag.StringVar(&address, "address", ":443", "server address")
	flag.StringVar(&host, "host", "localhost", "host/authority header")
	flag.IntVar(&threads, "threads", 1, "number of threads to start")
	flag.IntVar(&connections, "connections", 1, "number of connections to start")
	flag.IntVar(&debug, "debug", 0, "debug level")
	flag.Parse()
}

func main() {
	log.Printf("Starting %d connections in %d threads", connections, threads)
	var conn_per_thread int = connections / threads
	var rem int = connections % threads

	for i := 0; i < threads; i++ {
		inc := 0
		if rem > 0 {
			inc = 1
			rem--
		}
		go func(i int, inc int) {
			for j := 0; j < conn_per_thread + inc; j++ {
				ping_flood(i)
				//time.Sleep(1 * time.Second)
			}
		}(i, inc)
	}

	done := make(chan struct{})
	<-done
}

func ping_flood(cid int) {
	conf := &tls.Config{
         InsecureSkipVerify: true,
	 NextProtos: []string{"h2"},
	 MinVersion: tls.VersionTLS12,
	 MaxVersion: tls.VersionTLS12,
    	}
	conn, err := tls.DialWithDialer(&net.Dialer{Timeout:  2 * time.Second},
					"tcp", address, conf)
	if err != nil {
		if (debug > 0) {
			log.Printf("Connection error. Filtered?: %s\n", err)
		}
		return
	}

	_, err = conn.Write([]byte(http2.ClientPreface))
	if err != nil {
		panic(err)
	}

	framer := http2.NewFramer(conn, conn)

	initialSettings := []http2.Setting{
		{ID: http2.SettingEnablePush, Val: 0},
		{ID: http2.SettingInitialWindowSize, Val: transportDefaultStreamFlow},
		{ID: http2.SettingMaxFrameSize, Val: maxFrameSize},
		{ID: http2.SettingMaxHeaderListSize, Val: maxHeaderListSize},
		{ID: http2.SettingHeaderTableSize, Val: headerTableSize},
	}

	err = framer.WriteSettings(initialSettings...)
	if err != nil {
		panic(err)
	}

	go func() {
		var id uint64
		bs := make([]byte, 8)
		for {
			//_, err := readPrintFrame(cid, framer)
			frame, err := readPrintFrame(cid, framer)
			if err != nil {
				return
			}
			if frame.Header().Type == http2.FramePing {
				copy(bs, frame.(*http2.PingFrame).Data[:])
				i := binary.BigEndian.Uint64(bs)
				if id + 1 != i {
					log.Println("oop! id break:", i)
				}
				id = i
				//if id % 1000 == 0 {
				//	log.Println("ping ack id: ", id)
				//}
				//if id == 1000000 {
				//	log.Println("finished: ", id)
				//	break
				//}
			}
		}
	}()
	var id uint64
	var bs [8]byte
	for {
		id++
		binary.BigEndian.PutUint64(bs[:], id)
		//err = framer.WritePing(false, [8]byte{1,2,3,4})
		err = framer.WritePing(false, bs)
		if err != nil {
			log.Printf("[%d] Read: %s\n", cid, err)
			return
		}
		//if id % 1000 == 0 {
		//	log.Println("ping id: ", id)
		//}
		//if id == 100000000 {
		//	log.Println("stop ping: ", id)
		//	break
		//}
		//time.Sleep(10 * time.Microsecond)
	}
}

func readPrintFrame(cid int, framer *http2.Framer) (http2.Frame, error) {
	frame, err := framer.ReadFrame()
	if err != nil {
		log.Printf("[%d] Error reading %+v. Reason: %s\n", cid, frame, err)
		return frame, err
	}

	if debug > 1 {
		log.Printf("[%d] Read frame %+v\n", cid, frame)
	}

	if frame.Header().Type == http2.FrameGoAway {
		log.Println(frame.(*http2.GoAwayFrame).ErrCode)
	}

	return frame, nil
}

tempesta-ping.conf

listen 443 proto=h2;

cache_purge_acl 127.0.0.1;
frang_limits {
   http_strict_host_checking false;
   http_header_cnt 500;
   http_body_len 10485760; #10MB
}

srv_group default {
    server 127.0.0.1:8080 conns_n=1;
}

tls_certificate /home/kingluo/tempesta/server.crt;
tls_certificate_key /home/kingluo/tempesta/server.key;
tls_match_any_server_name;

vhost debian {
    resp_hdr_set Strict-Transport-Security "max-age=31536000; includeSubDomains";
    proxy_pass default;
}

cache 2;
cache_fulfill * *;
block_action attack reply;
block_action error reply;

http_chain {
  -> debian;
}
keepalive_timeout 600000;