second-transfer (empty) → 0.1.0.0
raw patch · 18 files changed
+3360/−0 lines, 18 filesdep +basedep +base16-bytestringdep +binarysetup-changed
Dependencies added: base, base16-bytestring, binary, bytestring, conduit, containers, hashable, hashtables, hslogger, http2, lens, network, network-uri, second-transfer, text, transformers
Files
- LICENSE +30/−0
- README.md +43/−0
- Setup.hs +2/−0
- cbits/tlsinc.c +785/−0
- hs-src/SecondTransfer.hs +141/−0
- hs-src/SecondTransfer/Http2.hs +5/−0
- hs-src/SecondTransfer/Http2/Framer.hs +430/−0
- hs-src/SecondTransfer/Http2/MakeAttendant.hs +28/−0
- hs-src/SecondTransfer/Http2/Session.hs +764/−0
- hs-src/SecondTransfer/MainLoop.hs +26/−0
- hs-src/SecondTransfer/MainLoop/CoherentWorker.hs +101/−0
- hs-src/SecondTransfer/MainLoop/Framer.hs +78/−0
- hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs +350/−0
- hs-src/SecondTransfer/MainLoop/PushPullType.hs +71/−0
- hs-src/SecondTransfer/MainLoop/Tokens.hs +150/−0
- hs-src/SecondTransfer/Utils.hs +153/−0
- second-transfer.cabal +159/−0
- tests/tests-hs-src/compiling_ok.hs +44/−0
+ LICENSE view
@@ -0,0 +1,30 @@+Copyright (c) 2015, Alcides Viamontes Esquivel++All rights reserved.++Redistribution and use in source and binary forms, with or without+modification, are permitted provided that the following conditions are met:++ * Redistributions of source code must retain the above copyright+ notice, this list of conditions and the following disclaimer.++ * Redistributions in binary form must reproduce the above+ copyright notice, this list of conditions and the following+ disclaimer in the documentation and/or other materials provided+ with the distribution.++ * Neither the name of Alcides Viamontes Esquivel nor the names of other+ contributors may be used to endorse or promote products derived+ from this software without specific prior written permission.++THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS+"AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT+LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR+A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT+OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,+SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT+LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,+DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY+THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT+(INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE+OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ README.md view
@@ -0,0 +1,43 @@+ +Developer README+================++Introduction+------------++This is an early-stage and very experimental library to create HTTP/2 servers+using Haskell. ++To see the package docs, please check the Hackage page or +the file hs-src/SecondTransfer.hs.++Building and installing+-----------------------++You need Haskell GHC compiler installed (version 7.8.3 at least). You also +need OpenSSL 1.0.2, since the ALPN feature and some very recent cypher-suites+are needed by HTTP/2. In this source distribution, I have set them to live in the +directory `/opt/openssl-1.0.2`, but you should be able to +alter the options using `cabal configure`. This package uses Haskell's foreign function +interface to interface with OpenSSL.++Provided that you have all the dependencies, you should be able to just do:++ $ cabal install second-transfer++Example+-------++There is a very basic example at `tests/tests-hs-src/compiling_ok.hs`. ++Roadmap+-------++Done:++- Version 0.1: Having something that can run. No unit-testing, nothing + fancy. ++Pending:++- Version 0.2: Have some unit tests. A minimal amount of them
+ Setup.hs view
@@ -0,0 +1,2 @@+import Distribution.Simple+main = defaultMain
+ cbits/tlsinc.c view
@@ -0,0 +1,785 @@++#define _BSD_SOURCE 1+#define _XOPEN_SOURCE 1 /* or any value < 500 */+#define _POSIX_C_SOURCE 1++++#include <sys/types.h>+#include <signal.h>+#include <sys/socket.h>+#include <netinet/in.h>+#include <netdb.h>+#include <stdio.h>+#include <string.h>+#include <stdlib.h>+#include <unistd.h>+#include <errno.h>+#include <pthread.h>+#include <bits/sigthread.h>+#include <sys/time.h>+#include <sys/types.h>++#include <openssl/rand.h>+#include <openssl/ssl.h>+#include <openssl/err.h>++// Simple structure to keep track of the handle, and+// of what needs to be freed later.+typedef struct {+ int socket;+ SSL_CTX *sslContext;+ char* protocol_list;+ int protocol_list_length;+} connection_t;++// Simple structure representing a session here+typedef struct {+ int socket;+ SSL *sslHandle;+ // Protocol index selected during negotiation....+ int protocol_index;+} wired_session_t;++// This is the public API of this server...+// No arguments, all are wired here somewhere... for now+connection_t* make_connection();+// Call when you are done +void close_connection(connection_t* conn);+// Wait for the next one...+#define ALL_OK 0 +#define BAD_HAPPENED 1+#define TIMEOUT_REACHED 3+// This is also a failed IO with SSL, but this one may be quite+// natural and we want to handle it differently+#define TRANSPORT_CLOSED 2+int wait_for_connection(connection_t* conn, int microseconds, wired_session_t** wired_session);+int send_data(wired_session_t* ws, char* buffer, int buffer_size);+int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd);+int get_selected_protocol(wired_session_t* ws){ return ws->protocol_index; }+void dispose_wired_session(wired_session_t* ws);+static int thread_setup(void);+++////////////////////////////////////////////////////////////////////////++ +static int threads_are_up = 0;++// Leaky implementation now +void close_connection(connection_t* conn)+{+ if (conn->socket)+ {+ close (conn->socket);+ conn->socket = 0;+ }++ if (conn->sslContext)+ {+ SSL_CTX_free(conn->sslContext);+ conn->sslContext = 0;+ }++ free(conn);+}++// For this example, we'll be testing on openssl.org++// Wired-in data that I will need to use?++#define CERTIFICATE_PLACE "/home/alcides/projects/rede/mimic/config/servercert.pem"+#define PRIVKEY_PLACE "/home/alcides/projects/rede/mimic/config/privkey.pem"+// This should bind to local interface+#define SERVER "www.httpdos.com"+// Grr+#define PORT 1060++// Establish a regular tcp connection+static int tcpStart (char* hostname, int portno, int* errorh)+{+ int error, handle;+ *errorh = 0;+ struct hostent *host;+ struct sockaddr_in server;++ bzero((char *) &server, sizeof(server));+ ++ host = gethostbyname (hostname);+ handle = socket (AF_INET, SOCK_STREAM, 0);+ int one = 1;+ setsockopt(handle, SOL_SOCKET, SO_REUSEADDR, &one, sizeof(one));+ if (handle == -1)+ {+ perror ("Socket");+ handle = 0;+ *errorh = BAD_HAPPENED;+ }+ else+ {+ server.sin_family = AF_INET;+ server.sin_port = htons (portno);+ server.sin_addr = *((struct in_addr *) host->h_addr);+ bzero (&(server.sin_zero), 8);++ error = bind (handle, (struct sockaddr *) &server,+ sizeof (struct sockaddr));+ if (error == -1)+ {+ perror ("Bind");+ handle = 0;+ *errorh = BAD_HAPPENED;+ }+ else + {+ error = listen(handle, 5);+ if ( error != 0 )+ {+ perror("Listen");+ handle = 0;+ *errorh = BAD_HAPPENED;+ }+ }+ }++ return handle;+}++// Dh_Callback {{{+#ifndef HEADER_DH_H+#include <openssl/dh.h>+#endif+DH *get_dh2236()+{+ static unsigned char dh2236_p[]={+ 0x08,0x32,0x3B,0x6A,0xE8,0xEA,0x55,0xE3,0x0C,0xD9,0x95,0x3D,+ 0xE7,0x78,0xDA,0xF0,0x05,0x81,0x94,0x85,0x34,0x5C,0x26,0x5A,+ 0xAE,0x9E,0x31,0x8F,0x0C,0xCC,0xFC,0xF3,0x30,0x88,0x20,0x0B,+ 0x41,0xC3,0xC9,0xEC,0x13,0xA7,0xFF,0xDA,0xBB,0x11,0xFB,0x7B,+ 0x19,0x92,0x3B,0x61,0xDF,0x31,0xEF,0x2C,0x4A,0x5C,0xD1,0x59,+ 0x16,0x90,0xAC,0xFC,0xEA,0xFF,0xCB,0xC4,0x04,0x07,0x93,0x9D,+ 0xF0,0x70,0x0B,0xCB,0x3E,0x79,0x0D,0xEB,0xD3,0x3E,0x06,0x0B,+ 0x19,0xF2,0x97,0x1E,0xE1,0xBF,0xD2,0xBE,0x5E,0xA5,0xBD,0x5F,+ 0x73,0x2C,0x57,0x0C,0xB0,0x97,0x78,0xAA,0x60,0xDC,0x98,0x89,+ 0xAF,0xBF,0xCD,0x49,0x74,0x82,0x64,0x09,0x60,0x47,0xD0,0xD4,+ 0x5D,0x25,0xDC,0x38,0xAC,0x17,0xE7,0xA4,0x47,0x59,0x94,0xFE,+ 0xA8,0xAD,0x58,0xD0,0xD1,0x43,0x3C,0x20,0x2E,0x34,0xEE,0xA9,+ 0x0F,0x71,0x20,0x93,0xB7,0x8E,0xEC,0xB1,0x75,0xB9,0xE9,0x9A,+ 0xAB,0x73,0xBB,0x7F,0xA2,0x8F,0x11,0xDE,0x58,0x5E,0xB0,0x98,+ 0xD6,0x95,0x84,0x62,0x90,0x87,0x90,0x32,0x34,0xF7,0x61,0x28,+ 0xFF,0x17,0xD0,0x58,0x33,0xA6,0xC2,0xC6,0x58,0x65,0x1C,0x92,+ 0xF3,0xDF,0x4D,0xB3,0xB0,0xFD,0xD2,0x4A,0x97,0x1B,0xA7,0xD2,+ 0x7C,0x8D,0x8F,0x1F,0x69,0x98,0x54,0xD9,0x33,0x15,0xE5,0xEA,+ 0xBA,0xAA,0x31,0xB1,0x17,0x65,0x21,0xFA,0xC7,0x54,0xC0,0xE4,+ 0x72,0x3C,0x15,0x74,0x41,0xF2,0x9A,0xF6,0xD1,0x16,0x10,0x35,+ 0x4F,0x36,0xA6,0x23,0x67,0x67,0x89,0x65,0xDD,0x77,0x5D,0x8F,+ 0x69,0x49,0x22,0x24,0xB6,0xB8,0x45,0xCA,0x8F,0x35,0x71,0x45,+ 0x4C,0x65,0x38,0x11,0x21,0xD5,0x39,0x03,0x4D,0x05,0xBB,0x95,+ 0x43,0x58,0xBB,0xF3,+ };+ static unsigned char dh2236_g[]={+ 0x02,+ };+ DH *dh;++ if ((dh=DH_new()) == NULL) return(NULL);+ dh->p=BN_bin2bn(dh2236_p,sizeof(dh2236_p),NULL);+ dh->g=BN_bin2bn(dh2236_g,sizeof(dh2236_g),NULL);+ if ((dh->p == NULL) || (dh->g == NULL))+ { DH_free(dh); return(NULL); }+ return(dh);+}+DH *get_dh2048()+{+static unsigned char dh2048_p[]={+ 0xCE,0x28,0x14,0x1C,0xEF,0x22,0x1D,0x86,0xEA,0x10,0x00,0x50,+ 0x24,0x42,0x95,0xC3,0x07,0x5A,0x87,0xED,0x0F,0xC5,0xDC,0x0F,+ 0x5E,0x7E,0x69,0x25,0x85,0x90,0x39,0x60,0x1E,0x87,0x5C,0x4B,+ 0xAD,0xDF,0xA8,0xF4,0x9C,0xC6,0x2D,0x6A,0x2E,0x7C,0xD1,0x5C,+ 0xC5,0x1A,0x74,0xD3,0x9E,0xE1,0xBB,0x31,0xC4,0x23,0x2F,0x78,+ 0xF5,0x61,0x5C,0x09,0x7E,0x29,0x06,0xC5,0x07,0x50,0x32,0x80,+ 0x4A,0xBE,0x5F,0x7F,0x68,0x69,0x8F,0xB1,0x6D,0xA0,0x0C,0x7F,+ 0x9E,0x28,0x77,0x9D,0x4F,0xEA,0xB5,0x38,0xB3,0x72,0x8E,0x1D,+ 0x8D,0x1C,0x58,0x74,0x58,0xC4,0xDD,0x06,0x8E,0x80,0x91,0x36,+ 0x2D,0x42,0x3D,0xF0,0x11,0xEC,0xDF,0x02,0xFD,0x84,0x54,0x32,+ 0x90,0xFE,0x7C,0x74,0x3A,0x5F,0x87,0xBA,0xD3,0x21,0x0E,0xDD,+ 0xF4,0xB3,0xFD,0xF9,0x89,0x8F,0x96,0x59,0x90,0x74,0xE8,0x45,+ 0x5A,0x3A,0x7C,0x88,0x7F,0xD8,0x5F,0xD2,0x32,0x57,0x46,0x29,+ 0x6B,0xA1,0x0A,0x05,0x1E,0xB6,0x49,0x8A,0x68,0xB4,0xEE,0x84,+ 0x45,0xEF,0x56,0x7E,0x59,0x83,0x67,0x20,0x85,0x63,0x69,0x6B,+ 0x39,0xCA,0x24,0x46,0x68,0x51,0x94,0x9E,0x3E,0xB4,0x69,0x1F,+ 0x63,0x07,0x40,0x0E,0x84,0x8B,0x26,0x98,0xCE,0xE5,0x48,0x2C,+ 0xD8,0xF9,0x6A,0x3F,0x20,0xFA,0xFA,0xDC,0x4E,0xFA,0xBD,0xC3,+ 0x81,0x09,0x4D,0xCF,0x07,0xEF,0xE8,0x32,0x9B,0x63,0x32,0x5B,+ 0x06,0x59,0x4F,0xE7,0x5B,0xD2,0xFC,0x25,0xC6,0x2C,0xA3,0xE8,+ 0x05,0xE5,0x8D,0xC2,0x94,0x76,0x90,0x63,0x29,0xB4,0xEE,0x2D,+ 0xD6,0x50,0x83,0x8B,+ };+static unsigned char dh2048_g[]={+ 0x02,+ };+DH *dh;++if ((dh=DH_new()) == NULL) return(NULL);+dh->p=BN_bin2bn(dh2048_p,sizeof(dh2048_p),NULL);+dh->g=BN_bin2bn(dh2048_g,sizeof(dh2048_g),NULL);+if ((dh->p == NULL) || (dh->g == NULL))+ { DH_free(dh); return(NULL); }+return(dh);+}++DH *get_dh1024()+ {+ static unsigned char dh1024_p[]={+ 0x87,0xF4,0xE5,0x5A,0x1E,0x7F,0x98,0x83,0xFD,0x23,0x6A,0xF3,+ 0x8C,0xE8,0x2F,0x35,0x64,0x3D,0x13,0x34,0x5B,0x0A,0x52,0xEC,+ 0x0B,0x3A,0xBF,0x92,0xE4,0x67,0x14,0x47,0x86,0x55,0x2C,0x83,+ 0x1B,0xDD,0x0E,0xC8,0x2D,0x98,0x72,0x2A,0xB6,0x68,0xF1,0x32,+ 0xD8,0xBD,0x6B,0x17,0x23,0x46,0x08,0xB6,0x19,0x58,0x01,0x30,+ 0x32,0x68,0x60,0x78,0xD1,0x5B,0x5C,0x88,0x3F,0x20,0xD7,0xEB,+ 0xE8,0xD5,0x80,0xB0,0x23,0x75,0xAE,0x97,0x26,0xBD,0xA5,0x11,+ 0x7F,0x83,0x8C,0x21,0xD4,0x39,0x52,0xE0,0x25,0x1F,0x03,0xA9,+ 0x69,0xF8,0x07,0x4F,0x33,0x17,0x72,0xCC,0xEB,0xBD,0x77,0xD1,+ 0x9C,0x40,0xA6,0x55,0xE3,0x87,0x76,0x66,0x3B,0xE3,0xE4,0x6A,+ 0x4B,0x6B,0xF1,0x92,0x1A,0x3A,0x25,0xC3,+ };+ static unsigned char dh1024_g[]={+ 0x02,+ };+ DH *dh;++ if ((dh=DH_new()) == NULL) return(NULL);+ dh->p=BN_bin2bn(dh1024_p,sizeof(dh1024_p),NULL);+ dh->g=BN_bin2bn(dh1024_g,sizeof(dh1024_g),NULL);+ if ((dh->p == NULL) || (dh->g == NULL))+ { DH_free(dh); return(NULL); }+ return (dh);+}+++static DH *tmp_dh_callback(SSL *s, int is_export, int keylength);+++// Setup dh params +static void setup_dh_parms(SSL_CTX *sslContext)+{+ /* Set up ephemeral DH stuff */+ SSL_CTX_set_tmp_dh_callback(sslContext, tmp_dh_callback);+}++static DH *tmp_dh_callback(SSL *s, int is_export, int keylength)+{+ static DH *dh_2048 = NULL;+ static DH *dh_1024 = NULL;+ DH *dh_tmp=NULL;+ switch (keylength) {+ case 2048:+ if (!dh_2048)+ dh_2048 = get_dh2048();+ dh_tmp = dh_2048;+ break;+ case 1024:+ if (!dh_1024)+ dh_1024 = get_dh1024();+ dh_tmp = dh_1024;+ break;+ default:+ /* Generating a key on the fly is very costly, so use what is there */+ printf("Keylength %d \n", keylength);+ }+ return(dh_tmp);+}+// dh_callback }}}++++static int protocol_select (+ SSL *ssl,+ const unsigned char **out,+ unsigned char *outlen,+ const unsigned char *in,+ unsigned int inlen,+ void *arg)+{+ // Oh well, C is a verbose beast+ connection_t* conn = (connection_t*) arg;+ static char output[64];++ char* incursor = (char*) in;++ while (incursor < (char*)in + inlen )+ {+ // Got a protocol.... can I satisfy it?+ char sublen = *incursor;+ printf("offered prot %.*s \n", sublen, incursor+1);++ char* stored_cursor = conn->protocol_list;+ int sto_protocol = 0;++ while( stored_cursor < conn->protocol_list + conn->protocol_list_length)+ {+ char sublen2 = *stored_cursor;++ if (sublen != sublen2)+ {+ + } else {+ int cmpresult = strncmp( incursor + 1, stored_cursor + 1, sublen);+ if (cmpresult == 0)+ {+ // They are equal, choose this one...+ strncpy( output, stored_cursor+1, sublen);+ *outlen = sublen;+ *out = output;+++ return SSL_TLSEXT_ERR_OK;+ }+ }+ sto_protocol += 1;+ stored_cursor += (1+sublen2);+ }++ incursor += (1+sublen);+ }++ // I think this is what should be returned + return -1;+}+++int lookup_protocol(+ char* selected, int selected_len, + char* myprotocol_list, int mpl_len+ )+{+ char* incursor = selected;++ char* stored_cursor = myprotocol_list;+ int sto_protocol = 0;++ if (selected_len == 0)+ {+ // No protocol was selected+ return -2;+ }++ while( stored_cursor < myprotocol_list + mpl_len)+ {+ char sublen2 = *stored_cursor;++ if (sublen2 != selected_len)+ {+ + } else {+ int cmpresult = strncmp( selected, stored_cursor + 1, sublen2);+ if (cmpresult == 0)+ {+ // They are equal, choose this one...+ return sto_protocol;+ }+ }+ sto_protocol += 1;+ stored_cursor += (1+sublen2);+ }++ // I think this is what should be returned + return -1;+}++static int ssl_servername_cb(SSL *s, int *ad, void *arg)+{+ // TODO: Use this for something .... although I'm not sure what...++ // tlsextctx *p = (tlsextctx *) arg;+ // const char *servername = SSL_get_servername(s, TLSEXT_NAMETYPE_host_name);+ // if (servername && p->biodebug)+ // BIO_printf(p->biodebug, "Hostname in TLS extension: \"%s\"\n",+ // servername);++ // if (!p->servername)+ // return SSL_TLSEXT_ERR_NOACK;++ // if (servername) {+ // if (strcasecmp(servername, p->servername))+ // return p->extension_error;+ // if (ctx2) {+ // BIO_printf(p->biodebug, "Switching server context.\n");+ // SSL_set_SSL_CTX(s, ctx2);+ // }+ // }+ return SSL_TLSEXT_ERR_OK;+}+++// Establish a connection using an SSL layer+static connection_t *sslStart (+ char* certificate_filename, char* privkey_filename, char* hostname, int portno,+ char* protocol_list, int protocol_list_length+ )+{+ int result;+ connection_t *c;++ if (! threads_are_up)+ {+ threads_are_up = 1;+ thread_setup();+ }++ c = malloc (sizeof (connection_t));+ c->sslContext = NULL;+ c->protocol_list = (char*) malloc(protocol_list_length);+ strncpy( c->protocol_list, protocol_list, protocol_list_length );+ c->protocol_list_length = protocol_list_length;+ c->socket = tcpStart (hostname, portno, &result);++ if ( result )+ {+ return 0;+ }++ if (c->socket)+ {+ // Register the error strings for libcrypto & libssl+ SSL_load_error_strings ();+ // Register the available ciphers and digests+ SSL_library_init ();++ // New context saying we are a server, and using SSL 2 or 3+ c->sslContext = SSL_CTX_new( TLSv1_2_server_method() );+ if (c->sslContext == NULL)+ {+ ERR_print_errors_fp (stderr);+ perror("Could not create context");+ return 0;+ }+++ // Now I set a few options....+ /*SSL_CTX_set_verify(c->sslContext, NULL );*/+ // const long flags = + // SSL_OP_NO_SSLv2 | SSL_OP_NO_SSLv3 | SSL_OP_NO_TLSv1 | SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION;+ const long flags = + SSL_OP_NO_SESSION_RESUMPTION_ON_RENEGOTIATION | SSL_OP_NO_COMPRESSION;+ SSL_CTX_set_options(c->sslContext, flags);+ setup_dh_parms( c->sslContext );+ SSL_CTX_set_ecdh_auto (c->sslContext, 1);++ // Give the impression that we are using SNI+ SSL_CTX_set_tlsext_servername_callback(c->sslContext, ssl_servername_cb);++++ // The only cipher supported by HTTP/2 ... sort of.+ result = SSL_CTX_set_cipher_list(c->sslContext, "ECDHE-RSA-AES128-GCM-SHA256");+ /*result = SSL_CTX_set_cipher_list(c->sslContext, "DEFAULT");*/++ // Be sure we are able to do ALPN+ // void SSL_CTX_set_alpn_select_cb(SSL_CTX *ctx,+ // int (*cb) (SSL *ssl,+ // const unsigned char **out,+ // unsigned char *outlen,+ // const unsigned char *in,+ // unsigned int inlen,+ // void *arg), void *arg);+ SSL_CTX_set_alpn_select_cb( c->sslContext, protocol_select, c);++ if ( result != 1 )+ {+ ERR_print_errors_fp (stderr);+ perror("Could not set cipher");+ return 0;+ }+ // The certificate (well-wired)+ SSL_CTX_use_certificate_file(c->sslContext, certificate_filename, SSL_FILETYPE_PEM);+ SSL_CTX_use_PrivateKey_file(c->sslContext, privkey_filename, SSL_FILETYPE_PEM);+ // Check the private key + result = SSL_CTX_check_private_key(c->sslContext);+ if ( result != 1)+ {+ perror("Check private key failed");+ return 0;+ }+ // //////////////////////////////////////////++ }+ else+ {+ perror ("Connect failed");+ }++ return c;+}+++connection_t* make_connection(char* certificate_filename, char* privkey_filename, char* hostname, int portno,+ char* my_protocol_list, int protocol_list_length+ )+{+ const int len = strlen(hostname);+ if ( len > 0 && hostname[len-1] == '\n' )+ { + printf("HELLO THERE. I'M A FLIMSY C FUNCTION AND I CAN NOT REMOVE THE END-OF-LINE IN THE PROVIDED HOST NAME WITHOUT A TON OF LINES. I'M GOING TO SEGFAULT NOW. BYE.\n");+ *( (int*) 0) = 42; // <-- This is the answer+ }+++ return sslStart(+ certificate_filename, privkey_filename, hostname, + portno, my_protocol_list, protocol_list_length); +}++int wait_for_connection(+ connection_t* c, + int microseconds,+ wired_session_t** wired_session)+{+ int clilen, newsockfd;+ // Don't return anything if there's a failure...+ *wired_session = 0;++ // Wait for a connection+ struct sockaddr_in cli_addr;+ clilen = sizeof(cli_addr);++ // Use select to get "interruptible" accepts + fd_set rfds;++ int can_go = 0;++ while( ! can_go )+ {+ FD_ZERO(&rfds);+ FD_SET(c->socket, &rfds);+ struct timeval tv;+ tv.tv_sec = microseconds / 1000000;+ tv.tv_usec = microseconds % 1000000;+++ int retval = select(FD_SETSIZE, &rfds, NULL, NULL, &tv);+++ if ( retval == -1 )+ {+ if ( errno == EINTR )+ {+ // printf(".");+ can_go = 0;+ } else {+ perror("select()");+ return BAD_HAPPENED;+ }+ } else if (retval > 0)+ {+ // We got data, just let it go...+ // printf("letitgo\n");+ can_go = 1;+ } else + {+ // We didn't get data, finish and terminate+ // printf("timeo\n");+ return TIMEOUT_REACHED;+ }++ }++ wired_session_t* result = (wired_session_t*) malloc(sizeof(wired_session_t) );+ if ( result == 0 )+ {+ perror("Malloc failed");+ return BAD_HAPPENED;+ }++ /* Accept actual connection from the client */+ newsockfd = accept(c->socket, (struct sockaddr *)&cli_addr, &clilen);++ if (newsockfd < 0)+ {+ perror("ERROR on accept");+ return BAD_HAPPENED;+ }+ + // Create an SSL struct for the connection++ result->socket = newsockfd;+ result->sslHandle = SSL_new (c->sslContext);+ if (result->sslHandle == NULL)+ {+ perror("No handle on SSL new");+ ERR_print_errors_fp (stderr);+ return BAD_HAPPENED;+ }++ // Connect the SSL struct to our connection+ if (!SSL_set_fd (result->sslHandle, result->socket))+ {+ perror("Could not associate");+ ERR_print_errors_fp (stderr);+ return BAD_HAPPENED;+ }++ // Initiate SSL handshake+ if (SSL_accept (result->sslHandle) != 1)+ {+ perror("Could not accept");+ ERR_print_errors_fp (stderr);+ return BAD_HAPPENED;+ }++ // After this one is okej, see which protocol was selected+ const unsigned char* out_protocol; unsigned pr_len;+ SSL_get0_alpn_selected(result->sslHandle, &out_protocol,+ &pr_len);+ // Wonder who is in charge of releasing that buffer...+ result -> protocol_index = lookup_protocol( + (char*)out_protocol, pr_len, + c->protocol_list, c->protocol_list_length);++ *wired_session = result;++ return ALL_OK;+}++// Disconnect & free connection struct+/*static void sslDisconnect (connection *c)*/+/*{*/+ /*if (c->socket)*/+ /*close (c->socket);*/+ /*if (c->sslHandle)*/+ /*{*/+ /*SSL_shutdown (c->sslHandle);*/+ /*SSL_free (c->sslHandle);*/+ /*}*/+ /*if (c->sslContext)*/+ /*SSL_CTX_free (c->sslContext);*/++ /*free (c);*/+/*}*/+++int send_data(wired_session_t* ws, char* buffer, int buffer_size)+{+ if ( buffer_size == 0 )+ {+ return ALL_OK;+ }+ if (ws)+ {+ int result = SSL_write( ws->sslHandle, buffer, buffer_size);+ if ( result > 0 )+ {+ return ALL_OK;+ } else {+ return BAD_HAPPENED;+ }+ } else + {+ return BAD_HAPPENED;+ }+}++int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd)+{+ int received=0, count = 0;+ char buffer[1024];++ // printf("Recvd entered\n");++ if (ws)+ {+ received = SSL_read (ws->sslHandle, inbuffer, buffer_size);+ }+ if ( received <= 0 )+ {+ ERR_print_errors_fp (stderr);+ return BAD_HAPPENED;+ }+ *data_recvd = received ;++ // printf("Recvd exited\n");++ return ALL_OK;+}++#define MUTEX_TYPE pthread_mutex_t+#define MUTEX_SETUP(x) pthread_mutex_init(&(x), NULL)+#define MUTEX_CLEANUP(x) pthread_mutex_destroy(&(x))+#define MUTEX_LOCK(x) pthread_mutex_lock(&(x))+#define MUTEX_UNLOCK(x) pthread_mutex_unlock(&(x))+#define THREAD_ID pthread_self( )+ + +void handle_error(const char *file, int lineno, const char *msg){+ fprintf(stderr, "** %s:%d %s\n", file, lineno, msg);+ ERR_print_errors_fp(stderr);+ /* exit(-1); */ + }+ +/* This array will store all of the mutexes available to OpenSSL. */ +static MUTEX_TYPE *mutex_buf= NULL;+ + +static void locking_function(int mode, int n, const char * file, int line)+{+ if (mode & CRYPTO_LOCK)+ MUTEX_LOCK(mutex_buf[n]);+ else+ MUTEX_UNLOCK(mutex_buf[n]);+}+ +static unsigned long id_function(void)+{+ return ((unsigned long)THREAD_ID);+}++void dispose_wired_session(wired_session_t* ws)+{+ if (ws == 0)+ return ;+ if ( ws-> sslHandle )+ {+ SSL_shutdown( ws->sslHandle );+ ws -> sslHandle = 0;+ }+ if (ws->socket)+ {+ close(ws->socket);+ ws -> socket = 0;+ }+ free(ws);+}+++int thread_setup(void)+{+ int i;++ // printf("Threads setup\n");+ + mutex_buf = malloc(CRYPTO_num_locks( ) * sizeof(MUTEX_TYPE));+ if (!mutex_buf)+ return 0;+ for (i = 0; i < CRYPTO_num_locks( ); i++)+ MUTEX_SETUP(mutex_buf[i]);+ CRYPTO_set_id_callback(id_function);+ CRYPTO_set_locking_callback(locking_function);+ return 1;+}+ +int thread_cleanup(void)+{+ int i;+ + if (!mutex_buf)+ return 0;+ CRYPTO_set_id_callback(NULL);+ CRYPTO_set_locking_callback(NULL);+ for (i = 0; i < CRYPTO_num_locks( ); i++)+ MUTEX_CLEANUP(mutex_buf[i]);+ free(mutex_buf);+ mutex_buf = NULL;+ return 1;+}+
+ hs-src/SecondTransfer.hs view
@@ -0,0 +1,141 @@+{-|+Module : SecondTransfer+Description : A library for implementing HTTP\/2 servers supporting streaming requests and responses.+Copyright : (c) Alcides Viamontes Esquivel, 2015+License : BSD+Maintainer : alcidesv@zunzun.se+Stability : experimental+Portability : POSIX++This library implements enough of the HTTP/2 to build +compliant HTTP/2 servers. The library++ * Is concurrent, meaning that you can use amazing Haskell lightweight threads to + process the requests. ++ * Obeys HTTP/2 flow control aspects.++ * And gives you freedom to (ab)use the HTTP/2 protocol in all the ways envisioned + by the standard. In particular you should be able to process streaming requests + (long uploads in POST or PUT requests) and to deliver streaming responses. You+ should even be able to do both simultaneously. ++Setting up TLS for HTTP/2 correctly is enough of a shore, so I have bundled here the+TLS setup logic. ++Frame encoding and decoding is done with +Kazu Yamamoto's <http://hackage.haskell.org/package/http2 http2> package. ++Here is how you create a very basic HTTP/2 webserver:++@+{-# LANGUAGE OverloadedStrings #-}+import SecondTransfer(+ CoherentWorker+ , DataAndConclusion+ , tlsServeWithALPN+ , http2Attendant+ )++import Data.Conduit+++saysHello :: 'DataAndConclusion'+saysHello = do + yield "Hello world!\\ns"+ -- No footers+ return []+++helloWorldWorker :: 'CoherentWorker'+helloWorldWorker request = return (+ [+ (":status", "200")+ ],+ [], -- No pushed streams+ saysHello+ )+++-- For this program to work, it should be run from the top of +-- the developement directory, so that it has access to the toy +-- certificates and keys defined there. +main = do + 'tlsServeWithALPN'+ "tests\/support\/servercert.pem" -- Server certificate+ "tests\/support\/privkey.pem" -- Certificate private key+ "127.0.0.1" -- On which interface to bind+ [+ ("h2-14", http2_attendant), -- Protocols present in the ALPN negotiation+ ("h2", http2_attendant) -- they may be slightly different, but for this + -- test it doesn't matter.+ ]+ 8000+ where + http2_attendant = http2Attendant helloWorldWorker+@++`CoherentWorker` is the basic callback function that you need to implement. +The callback is used to handle all requests to the server on a given negotiated ALPN +protocol. If you need routing functionality (and you most certainly will need it), you need+to build that functionality yourself or use one of the many Haskell libraries to that+end. ++The above program uses a test certificate by a fake certificate authority. The certificate+is valid for the server name ("authority", in HTTP\/2 lingo) www.httpdos.com. So, in order+for the above program to run, you probably need to add an alias to your \/etc\/hosts file. +You also need very up-to-date versions of OpenSSL (I'm using OpenSSL 1.0.2) to be compliant+with the cipher suites demanded by HTTP\/2. The easiest way to test the above program is using+a fairly recent version of <http://curl.haxx.se/ curl>. If everything is allright, +you should be able to do:++@+ $ curl -k --http2 https://www.httpdos.com:8000/+ Hello world!+@++-}+module SecondTransfer(++ -- * Types related to coherent workers+ --+ -- | A coherent worker is an abstraction that can dance at the + -- tune of HTTP/2. That is, it should be able to take+ -- headers request first, and then a source of data coming in the + -- request (for example, POST data). Even before exhausting the source, + -- the coherent worker can post the response headers, and then create + -- its source for the response data. A coherent worker can also present+ -- create streams to push to the client. + Headers+ , Request+ , Footers+ , CoherentWorker+ , PrincipalStream+ , PushedStreams+ , PushedStream+ , DataAndConclusion+ , InputDataStream+ , FinalizationHeaders+ + -- * Basic utilities for HTTP/2 servers+ ,Attendant+ ,PullAction+ ,PushAction+ ,CloseAction+ ,http2Attendant+ ,IOProblem+ ,GenericIOProblem+ -- * High level OpenSSL functions. + -- + -- | Use these functions to create your TLS-compliant + -- HTTP/2 server in a snap.+ ,tlsServeWithALPN+ ,tlsServeWithALPNAndFinishOnRequest++ ,TLSLayerGenericProblem(..)+ ,FinishRequest(..)+ ) where ++import SecondTransfer.MainLoop.CoherentWorker +import SecondTransfer.MainLoop+import SecondTransfer.Http2.MakeAttendant(http2Attendant)
+ hs-src/SecondTransfer/Http2.hs view
@@ -0,0 +1,5 @@+module SecondTransfer.Http2(+ http2Attendant+ ) where++import SecondTransfer.Http2.MakeAttendant(http2Attendant)
+ hs-src/SecondTransfer/Http2/Framer.hs view
@@ -0,0 +1,430 @@+-- The framer has two functions: to convert bytes to Frames and the other way around,+-- and two keep track of flow-control quotas. +{-# LANGUAGE OverloadedStrings, StandaloneDeriving, FlexibleInstances, + DeriveDataTypeable, TemplateHaskell #-}+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.Http2.Framer (+ wrapSession,++ -- Not needed anywhere, but supress the warning about unneeded symbol+ closeAction+ ) where ++++import Control.Concurrent+import Control.Exception+import qualified Control.Exception as E+import Control.Monad.IO.Class (liftIO)+import Control.Monad.Trans.Class (lift)+import Control.Monad.Trans.Reader +import qualified Control.Lens as L+import Control.Lens (view)+import Data.Binary (decode)+import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as LB+import Data.Conduit+import Data.Typeable (Typeable)+import Data.Foldable (find)++import qualified Network.HTTP2 as NH2+++import qualified Data.HashTable.IO as H++import SecondTransfer.Http2.Session+import SecondTransfer.MainLoop.CoherentWorker (CoherentWorker)+import qualified SecondTransfer.MainLoop.Framer as F+import SecondTransfer.MainLoop.PushPullType (Attendant, PullAction,+ PushAction, CloseAction)+import SecondTransfer.Utils (Word24, word24ToInt)+++http2PrefixLength :: Int+http2PrefixLength = B.length NH2.connectionPreface+++data BadPrefixException = BadPrefixException + deriving (Show, Typeable)++instance Exception BadPrefixException++++-- Let's do flow control here here .... ++type HashTable k v = H.CuckooHashTable k v+++type GlobalStreamId = Int+++data FlowControlCommand = + AddBytes_FCM Int ++-- A hashtable from stream id to channel of availabiliy increases+type Stream2AvailSpace = HashTable GlobalStreamId (Chan FlowControlCommand)+++data CanOutput = CanOutput+++data NoHeadersInChannel = NoHeadersInChannel+++data FramerSessionData = FramerSessionData {+ _stream2flow :: Stream2AvailSpace+ , _stream2outputBytes :: HashTable GlobalStreamId (Chan LB.ByteString)+ , _defaultStreamWindow :: MVar Int++ , _canOutput :: MVar CanOutput+ , _noHeadersInChannel :: MVar NoHeadersInChannel+ , _pushAction :: PushAction+ , _closeAction :: CloseAction+ }+++L.makeLenses ''FramerSessionData+++type FramerSession = ReaderT FramerSessionData IO+++wrapSession :: CoherentWorker -> Attendant+wrapSession coherent_worker push_action pull_action close_action = do++ let session_start = SessionStartData {}++ (session_input, session_output) <- http2Session coherent_worker session_start++ -- TODO : Add type annotations....+ s2f <- H.new + s2o <- H.new + default_stream_size_mvar <- newMVar 65536+ can_output <- newMVar CanOutput+ no_headers_in_channel <- newMVar NoHeadersInChannel+++ -- We need some shared state + let framer_session_data = FramerSessionData {+ _stream2flow = s2f+ ,_stream2outputBytes = s2o + ,_defaultStreamWindow = default_stream_size_mvar+ ,_canOutput = can_output + ,_noHeadersInChannel = no_headers_in_channel+ ,_pushAction = push_action+ ,_closeAction = close_action+ }++ forkIO $ close_on_error $ runReaderT (inputGatherer pull_action session_input ) framer_session_data + forkIO $ close_on_error $ runReaderT (outputGatherer session_output ) framer_session_data ++ return ()++ where + close_on_error comp = E.finally comp close_action+++http2FrameLength :: F.LengthCallback+http2FrameLength bs | (B.length bs) >= 3 = let+ word24 = decode input_as_lbs :: Word24+ input_as_lbs = LB.fromStrict bs+ in + Just $ (word24ToInt word24) + 9 -- Nine bytes that the frame header always uses+http2FrameLength _ = Nothing+++addCapacity :: + GlobalStreamId ->+ Int -> + FramerSession ()+addCapacity stream_id delta_cap = do ++ if stream_id == 0 + then + -- TODO: Add session flow control+ return ()+ else do+ table <- view stream2flow + val <- liftIO $ H.lookup table stream_id+ case val of + Nothing -> do+ -- ??+ liftIO $ putStrLn $ "Tried to update window of unexistent stream (creating): " ++ (show stream_id)+ (_, command_chan) <- startStreamOutputQueue stream_id + -- And try again+ liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap+++ Just command_chan -> do+ liftIO $ writeChan command_chan $ AddBytes_FCM delta_cap+ ++-- This works by pulling bytes from the input side of the pipeline and converting them to frames.+-- The frames are then put in the SessionInput. In the other end of the SessionInput they can be +-- interpreted according to their HTTP/2 meaning. +-- +-- This function also does part of the flow control: it registers WindowUpdate frames and triggers+-- quota updates on the streams. +inputGatherer :: PullAction -> SessionInput -> FramerSession ()+inputGatherer pull_action session_input = do + -- We can start by reading off the prefix....+ (prefix, remaining) <- liftIO $ F.readLength http2PrefixLength pull_action++ if prefix /= NH2.connectionPreface + then + liftIO $ throwIO BadPrefixException+ else + return ()++ let + source::Source FramerSession B.ByteString+ source = transPipe liftIO $ F.readNextChunk http2FrameLength remaining pull_action+ ( source $$ consume)+ where ++ consume :: Sink B.ByteString FramerSession ()+ consume = do + maybe_bytes <- await + -- Deserialize++ case maybe_bytes of ++ Just bytes -> do+ let + error_or_frame = NH2.decodeFrame some_settings bytes+ -- TODO: See how we can change these....+ some_settings = NH2.defaultSettings++ case error_or_frame of ++ Left some_error -> do + liftIO $ putStrLn $ "Got an error: " ++ (show some_error)+++ Right (NH2.Frame (NH2.FrameHeader _ _ stream_id) (NH2.WindowUpdateFrame credit) ) -> do + -- Bookkeep the increase on bytes on that stream+ -- liftIO $ putStrLn $ "Extra capacity for stream " ++ (show stream_id)+ lift $ addCapacity (NH2.fromStreamIdentifier stream_id) (fromIntegral credit)+ return ()+++ Right frame@(NH2.Frame _ (NH2.SettingsFrame settings_list) ) -> do + -- Increase all the stuff....+ case find (\(i,_) -> i == NH2.SettingsInitialWindowSize) settings_list of ++ Just (_, new_default_stream_size) -> do + old_default_stream_size_mvar <- view defaultStreamWindow+ old_default_stream_size <- liftIO $ takeMVar old_default_stream_size_mvar+ let general_delta = new_default_stream_size - old_default_stream_size+ stream_to_flow <- view stream2flow+ -- Add capacity to everybody's windows+ liftIO $ + H.mapM_ (+ \ (k,v) -> if k /=0 + then writeChan v (AddBytes_FCM general_delta) + else return () )+ stream_to_flow+++ -- And set a new value + liftIO $ putMVar old_default_stream_size_mvar new_default_stream_size+++ Nothing -> + return ()++ -- And send the frame down to the session+ liftIO $ sendFrameToSession session_input frame+++ Right a_frame -> do + liftIO $ sendFrameToSession session_input a_frame++ -- tail recursion: go again...+ consume ++ Nothing -> + -- We may as well exit this thread+ return ()+++outputGatherer :: SessionOutput -> FramerSession ()+outputGatherer session_output = do ++ -- We start by sending a settings frame + pushFrame + (NH2.EncodeInfo NH2.defaultFlags (NH2.toStreamIdentifier 0) Nothing)+ (NH2.SettingsFrame [])++ loopPart++ where +++ dataForFrame p1 p2 = + LB.fromStrict $ NH2.encodeFrame p1 p2++ loopPart :: FramerSession ()+ loopPart = do ++ command_or_frame <- liftIO $ getFrameFromSession session_output++ case command_or_frame of ++ Left cmd -> do + -- TODO: This is just a quickie behavior I dropped + -- here, semantics need to be different probably.+ liftIO $ putStrLn $ "Received a command... terminating " ++ (show cmd)+++ Right ( p1@(NH2.EncodeInfo _ stream_idii _), p2@(NH2.DataFrame _) ) -> do+ -- This frame is flow-controlled... I may be unable to send this frame in+ -- some circumstances... + let stream_id = NH2.fromStreamIdentifier stream_idii+ s2o <- view stream2outputBytes+ lookup_result <- liftIO $ H.lookup s2o stream_id + stream_bytes_chan <- case lookup_result of ++ Nothing -> do + (bc, _) <- startStreamOutputQueue stream_id+ return bc++ Just bytes_chan -> return bytes_chan ++ liftIO $ writeChan stream_bytes_chan $ dataForFrame p1 p2++ loopPart+++ Right (p1, p2@(NH2.HeadersFrame _ _) ) -> do+ handleHeadersOfStream p1 p2+ + loopPart+++ Right (p1, p2@(NH2.ContinuationFrame _) ) -> do+ handleHeadersOfStream p1 p2++ loopPart++ Right (p1, p2) -> do + -- Most other frames go right away... as long as no headers are in process...+ no_headers <- view noHeadersInChannel+ liftIO $ takeMVar no_headers+ pushFrame p1 p2 + liftIO $ putMVar no_headers NoHeadersInChannel+ + loopPart+++startStreamOutputQueue :: Int -> FramerSession (Chan LB.ByteString, Chan FlowControlCommand)+startStreamOutputQueue stream_id = do+ -- New thread for handling outputs of this stream is needed+ bytes_chan <- liftIO newChan + command_chan <- liftIO newChan + s2o <- view stream2outputBytes+ liftIO $ H.insert s2o stream_id bytes_chan + s2c <- view stream2flow+ liftIO $ H.insert s2c stream_id command_chan + initial_cap_mvar <- view defaultStreamWindow+ initial_cap <- liftIO $ readMVar initial_cap_mvar++ -- And don't forget the thread itself+ read_state <- ask + liftIO $ forkIO $ runReaderT+ (flowControlOutput stream_id initial_cap "" command_chan bytes_chan)+ read_state++ return (bytes_chan , command_chan)+++-- This works in the output side of the HTTP/2 framing session, and it acts as a +-- semaphore ensuring that headers are output without any interleaved frames. +handleHeadersOfStream :: NH2.EncodeInfo -> NH2.FramePayload -> FramerSession ()+handleHeadersOfStream p1@(NH2.EncodeInfo _ _ _) frame_payload+ | (frameIsHeaderOfStream frame_payload) && (not $ frameEndsHeaders p1 frame_payload) = do+ -- Take it + no_headers <- view noHeadersInChannel+ liftIO $ takeMVar no_headers+ pushFrame p1 frame_payload + -- DONT PUT THE MvAR HERE ++ | (frameIsHeaderOfStream frame_payload) && (frameEndsHeaders p1 frame_payload) = do+ no_headers <- view noHeadersInChannel+ liftIO $ takeMVar no_headers+ pushFrame p1 frame_payload + -- Since we finish.... + liftIO $ putMVar no_headers NoHeadersInChannel++ | frameEndsHeaders p1 frame_payload = do + -- I can only get here for a continuation frame after something else that is a headers+ no_headers <- view noHeadersInChannel+ result <- liftIO $ tryPutMVar no_headers NoHeadersInChannel+ liftIO $ putStrLn $ "Could put MVAR (yes must be): " ++ (show result)+++frameIsHeaderOfStream :: NH2.FramePayload -> Bool+frameIsHeaderOfStream (NH2.HeadersFrame _ _ )+ = True+frameIsHeaderOfStream _ + = False +++frameEndsHeaders :: NH2.EncodeInfo -> NH2.FramePayload -> Bool +frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.HeadersFrame _ _) = NH2.testEndHeader flags+frameEndsHeaders (NH2.EncodeInfo flags _ _) (NH2.ContinuationFrame _) = NH2.testEndHeader flags+frameEndsHeaders _ _ = False+++pushFrame :: NH2.EncodeInfo+ -> NH2.FramePayload -> FramerSession ()+pushFrame p1 p2 = do+ let bs = LB.fromStrict $ NH2.encodeFrame p1 p2 + sendBytes bs+++sendBytes :: LB.ByteString -> FramerSession ()+sendBytes bs = do+ push_action <- view pushAction+ can_output <- view canOutput + liftIO $ do + bs `seq` takeMVar can_output+ push_action bs+ putMVar can_output CanOutput+++-- A thread in charge of doing flow control transmission+-- TODO: Do session flow control..... +flowControlOutput :: Int -> Int -> LB.ByteString -> (Chan FlowControlCommand) -> (Chan LB.ByteString) -> FramerSession ()+flowControlOutput stream_id capacity leftovers commands_chan bytes_chan = do + -- Get some bytes to send + ++ if leftovers == "" + then do+ -- Get more data (possibly block waiting for it)+ bytes_to_send <- liftIO $ readChan bytes_chan+ flowControlOutput stream_id capacity bytes_to_send commands_chan bytes_chan+ else do+ -- Length?+ let amount = fromIntegral $ ((LB.length leftovers) - 9)+ if amount <= capacity + then do+ -- I can send ... if no headers are in process....+ no_headers <- view noHeadersInChannel+ liftIO $ takeMVar no_headers+ sendBytes leftovers+ -- liftIO $ putStrLn $ "Sent flow-controlled data for " ++ (show stream_id)+ -- liftIO $ putStrLn $ "Capacity left " ++ (show (capacity - amount))+ liftIO $ putMVar no_headers NoHeadersInChannel+ -- and tail-invoke + flowControlOutput stream_id (capacity - amount) "" commands_chan bytes_chan+ else do+ -- I can not send because flow-control is full, wait for a command instead + liftIO $ putStrLn $ "Warning: channel flow-saturated " ++ (show stream_id)+ command <- liftIO $ readChan commands_chan+ case command of + AddBytes_FCM delta_cap -> do + -- liftIO $ putStrLn $ "Flow control delta_cap stream " ++ (show stream_id)+ flowControlOutput stream_id (capacity + delta_cap) leftovers commands_chan bytes_chan
+ hs-src/SecondTransfer/Http2/MakeAttendant.hs view
@@ -0,0 +1,28 @@+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.Http2.MakeAttendant (+ http2Attendant+ ) where+++import SecondTransfer.Http2.Framer (wrapSession)+import SecondTransfer.MainLoop.CoherentWorker+import SecondTransfer.MainLoop.PushPullType (+ --CloseAction,+ --PullAction, + --PushAction,+ Attendant+ )++-- | The type of this function is equivalent to:+-- +-- @ +-- http2Attendant :: CoherentWorker -> PushAction -> PullAction -> CloseAction -> IO ()+-- @+-- +-- Given a `CoherentWorker`, this function wraps it with flow control, multiplexing,+-- and state maintenance needed to run an HTTP/2 session. +http2Attendant :: CoherentWorker -> Attendant+http2Attendant coherent_worker push_action pull_action close_action = do + let + attendant = wrapSession coherent_worker+ attendant push_action pull_action close_action
+ hs-src/SecondTransfer/Http2/Session.hs view
@@ -0,0 +1,764 @@+-- Session: links frames to streams, and helps in ordering the header frames+-- so that they don't get mixed with header frames from other streams when +-- resources are being served concurrently.+{-# LANGUAGE FlexibleContexts, Rank2Types, TemplateHaskell #-}+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.Http2.Session(+ http2Session+ ,getFrameFromSession+ ,sendFrameToSession+ ,sendCommandToSession++ ,CoherentSession+ ,SessionInput(..)+ ,SessionInputCommand(..)+ ,SessionOutput(..)+ ,SessionStartData(..)+ ) where+++-- System grade utilities+import Control.Monad (forever)+import Control.Concurrent (forkIO, ThreadId)+import Control.Concurrent.Chan+import Control.Monad.IO.Class (liftIO)+import Control.Monad.Trans.Reader+import Control.Exception (throwTo)++import Data.Conduit+import Data.Conduit.List (foldMapM)+import qualified Data.ByteString as B+import Control.Concurrent.MVar+import qualified Data.IntSet as NS+import qualified Data.HashTable.IO as H++import Control.Lens++-- No framing layer here... let's use Kazu's Yamamoto library+import qualified Network.HTTP2 as NH2+import qualified Network.HPACK as HP++-- Logging utilities+import System.Log.Logger++-- Imports from other parts of the program+import SecondTransfer.MainLoop.CoherentWorker +import SecondTransfer.MainLoop.Tokens+import SecondTransfer.Utils (unfoldChannelAndSource)+++-- Unfortunately the frame encoding API of Network.HTTP2 is a bit difficult to +-- use :-( +type OutputFrame = (NH2.EncodeInfo, NH2.FramePayload)+type InputFrame = NH2.Frame+++useChunkLength :: Int +useChunkLength = 16384+++-- Singleton instance used for concurrency+data HeadersSent = HeadersSent +++-- Whatever a worker thread is going to need comes here.... +-- this is to make refactoring easier, but not strictly needed. +data WorkerThreadEnvironment = WorkerThreadEnvironment {+ -- What's the header stream id?+ _streamId :: GlobalStreamId++ -- A full block of headers can come here... the mvar in the middle should+ -- be populate to signal end of headers transmission. A thread will be suspended+ -- waiting for that+ , _headersOutput :: Chan (GlobalStreamId, MVar HeadersSent, Headers)++ -- And regular contents can come this way and thus be properly mixed+ -- with everything else.... for now... + ,_dataOutput :: Chan (GlobalStreamId, B.ByteString)++ ,_streamsCancelled_WTE :: MVar NS.IntSet++ }++makeLenses ''WorkerThreadEnvironment+++-- Basically a couple of channels ... +type Session = (SessionInput, SessionOutput)+++-- From outside, one can only write to this one ... the newtype is to enforce +-- this.+newtype SessionInput = SessionInput ( Chan (Either SessionInputCommand InputFrame) )+sendFrameToSession :: SessionInput -> InputFrame -> IO ()+sendFrameToSession (SessionInput chan) frame = writeChan chan $ Right frame++sendCommandToSession :: SessionInput -> SessionInputCommand -> IO ()+sendCommandToSession (SessionInput chan) command = writeChan chan $ Left command++-- From outside, one can only read from this one +newtype SessionOutput = SessionOutput ( Chan (Either SessionOutputCommand OutputFrame) )+getFrameFromSession :: SessionOutput -> IO (Either SessionOutputCommand OutputFrame) +getFrameFromSession (SessionOutput chan) = readChan chan+++-- Here is how we make a session +type SessionMaker = SessionStartData -> IO Session++++-- Here is how we make a session wrapping a CoherentWorker+type CoherentSession = CoherentWorker -> SessionMaker +++type HashTable k v = H.CuckooHashTable k v+++-- Blaze builder could be more proper here... +type Stream2HeaderBlockFragment = HashTable GlobalStreamId B.ByteString+++type WorkerMonad = ReaderT WorkerThreadEnvironment IO +++-- Have to figure out which are these...but I would expect to have things+-- like unexpected aborts here in this type.+data SessionInputCommand = + CancelSession_SIC+ deriving Show +++-- temporary+data SessionOutputCommand = + CancelSession_SOC+ deriving Show+++-- TODO: Put here information needed for the session to work+data SessionStartData = SessionStartData {+ + }+++makeLenses ''SessionStartData+++data PostInputMechanism = PostInputMechanism (Chan (Maybe B.ByteString), InputDataStream)+++-- NH2.Frame != Frame+data SessionData = SessionData {+ _sessionInput :: Chan (Either SessionInputCommand InputFrame)++ -- We need to lock this channel occassionally so that we can order multiple + -- header frames properly.... + ,_sessionOutput :: MVar (Chan (Either SessionOutputCommand OutputFrame))++ -- Use to encode + ,_toEncodeHeaders :: MVar HP.DynamicTable+ -- And used to decode+ ,_toDecodeHeaders :: MVar HP.DynamicTable++ -- Used for decoding the headers+ ,_stream2HeaderBlockFragment :: Stream2HeaderBlockFragment++ -- Used for worker threads... this is actually a pre-filled template+ -- I make copies of it in different contexts, and as needed. + ,_forWorkerThread :: WorkerThreadEnvironment++ ,_coherentWorker :: CoherentWorker++ -- Some streams may be cancelled + ,_streamsCancelled :: MVar NS.IntSet++ -- Data input mechanism corresponding to some threads+ ,_stream2PostInputMechanism :: HashTable Int PostInputMechanism ++ -- Worker thread register. This is a dictionary from stream id to + -- the ThreadId of the thread with the worker thread. I use this to + -- raise asynchronous exceptions in the worker thread if the stream + -- is cancelled by the client+ ,_stream2WorkerThread :: HashTable Int ThreadId+ }+++makeLenses ''SessionData+++-- v- {headers table size comes here!!}+http2Session :: CoherentWorker -> SessionStartData -> IO Session+http2Session coherent_worker _ = do + session_input <- newChan+ session_output <- newChan+ session_output_mvar <- newMVar session_output+++ -- For incremental construction of headers...+ stream_request_headers <- H.new :: IO Stream2HeaderBlockFragment++ -- Warning: we should find a way of coping with different table sizes.+ decode_headers_table <- HP.newDynamicTableForDecoding 4096+ decode_headers_table_mvar <- newMVar decode_headers_table++ encode_headers_table <- HP.newDynamicTableForEncoding 4096+ encode_headers_table_mvar <- newMVar encode_headers_table++ -- These ones need independent threads taking care of sending stuff+ -- their way... + headers_output <- newChan :: IO (Chan (GlobalStreamId, MVar HeadersSent, Headers))+ data_output <- newChan :: IO (Chan (GlobalStreamId,B.ByteString))++ stream2postinputmechanism <- H.new + stream2workerthread <- H.new++ -- What about stream cancellation?+ cancelled_streams_mvar <- newMVar $ NS.empty :: IO (MVar NS.IntSet)++ let for_worker_thread = WorkerThreadEnvironment {+ _streamId = error "NotInitialized" + ,_headersOutput = headers_output+ ,_dataOutput = data_output+ ,_streamsCancelled_WTE = cancelled_streams_mvar+ }++ let session_data = SessionData {+ _sessionInput = session_input + ,_sessionOutput = session_output_mvar+ ,_toDecodeHeaders = decode_headers_table_mvar+ ,_toEncodeHeaders = encode_headers_table_mvar+ ,_stream2HeaderBlockFragment = stream_request_headers+ ,_forWorkerThread = for_worker_thread+ ,_coherentWorker = coherent_worker+ ,_streamsCancelled = cancelled_streams_mvar+ ,_stream2PostInputMechanism = stream2postinputmechanism+ ,_stream2WorkerThread = stream2workerthread+ }++ -- Create an input thread that decodes frames...+ forkIO $ runReaderT sessionInputThread session_data+ + -- Create a thread that captures headers and sends them down the tube + forkIO $ runReaderT (headersOutputThread headers_output session_output_mvar) session_data++ -- Create a thread that captures data and sends it down the tube+ forkIO $ dataOutputThread data_output session_output_mvar++ -- The two previous thread fill the session_output argument below (they write to it)+ -- the session machinery in the other end is in charge of sending that data through the + -- socket.+ ++ return ( (SessionInput session_input),+ (SessionOutput session_output) )++++sessionInputThread :: ReaderT SessionData IO ()+sessionInputThread = do + liftIO $ debugM "HTTP2.Session" "Entering sessionInputThread"++ -- This is an introductory and declarative block... all of this is tail-executed+ -- every time that a packet needs to be processed. It may be a good idea to abstract+ -- these values in a closure... + session_input <- view sessionInput + + decode_headers_table_mvar <- view toDecodeHeaders + stream_request_headers <- view stream2HeaderBlockFragment+ cancelled_streams_mvar <- view streamsCancelled+ coherent_worker <- view coherentWorker++ for_worker_thread_uns <- view forWorkerThread+ stream2workerthread <- view stream2WorkerThread++ input <- liftIO $ readChan session_input++ liftIO $ debugM "HTTP2.Session" $ "Got a frame or a command: " ++ (show input)++ case input of ++ Left CancelSession_SIC -> do + -- Good place to tear down worker threads... Let the rest of the finalization+ -- to somebody else....+ liftIO $ do + H.mapM_+ (\ (_, thread_id) -> do+ throwTo thread_id StreamCancelledException+ infoM "HTTP2.Session" $ "Stream successfully interrupted"+ )+ stream2workerthread++ -- We do not continue here, but instead let it finish+ return ()++ Right frame | Just (stream_id, bytes) <- frameIsHeaderOfStream frame -> do + -- Just append the frames to streamRequestHeaders+ appendHeaderFragmentBlock stream_id bytes++ if frameEndsHeaders frame then + do+ -- Let's decode the headers+ let for_worker_thread = set streamId stream_id for_worker_thread_uns + headers_bytes <- getHeaderBytes stream_id+ dyn_table <- liftIO $ takeMVar decode_headers_table_mvar+ (new_table, header_list ) <- liftIO $ HP.decodeHeader dyn_table headers_bytes+ -- Good moment to remove the headers from the table.... we don't want a space+ -- leak here + liftIO $ H.delete stream_request_headers stream_id+ liftIO $ putMVar decode_headers_table_mvar new_table++ -- If the headers end the request.... + post_data_source <- if not (frameEndsStream frame)+ then do + + mechanism <- createMechanismForStream stream_id + let source = postDataSourceFromMechanism mechanism+ return $ Just source+ else do + -- liftIO $ putStrLn "Headers end reqeust"+ return Nothing+++ -- I'm clear to start the worker, in its own thread+ -- !!+ liftIO $ do + thread_id <- forkIO $ runReaderT + (workerThread (header_list, post_data_source) coherent_worker)+ for_worker_thread + H.insert stream2workerthread stream_id thread_id++ return ()+ else + -- Frame doesn't end the headers... it was added before... so+ -- probably do nothing + return ()+ + continue ++ Right frame@(NH2.Frame _ (NH2.RSTStreamFrame error_code_id)) -> do+ let stream_id = streamIdFromFrame frame+ liftIO $ do + infoM "HTTP2.Session" $ "Stream reset: " ++ (show error_code_id)+ cancelled_streams <- takeMVar cancelled_streams_mvar+ infoM "HTTP2.Session" $ "Cancelled stream was: " ++ (show stream_id)+ putMVar cancelled_streams_mvar $ NS.insert stream_id cancelled_streams+ maybe_thread_id <- H.lookup stream2workerthread stream_id+ case maybe_thread_id of + Nothing -> + errorM "HTTP2.Session" $ "Attention: could not find stream " ++ (show stream_id) ++ ("in threads register")++ Just thread_id -> do+ throwTo thread_id StreamCancelledException+ infoM "HTTP2.Session" $ "Stream successfully interrupted"++ continue +++ Right frame@(NH2.Frame (NH2.FrameHeader _ _ nh2_stream_id) (NH2.DataFrame somebytes)) -> do + -- So I got data to process+ -- TODO: Handle end of stream+ let stream_id = NH2.fromStreamIdentifier nh2_stream_id + streamWorkerSendData stream_id somebytes+ -- After that data has been received and forwarded downstream, we can issue a windows update+ --+ -- TODO: We can use wider frames to avoid congestion...+ -- .... and we can also be more compositional with these short bursts of data....+ sendOutFrame+ (NH2.EncodeInfo+ NH2.defaultFlags+ nh2_stream_id+ Nothing+ )+ (NH2.WindowUpdateFrame+ (fromIntegral (B.length somebytes))+ )+ sendOutFrame+ (NH2.EncodeInfo+ NH2.defaultFlags+ (NH2.toStreamIdentifier 0)+ Nothing+ )+ (NH2.WindowUpdateFrame+ (fromIntegral (B.length somebytes))+ ) ++ if frameEndsStream frame + then do + -- Good place to close the source ... + closePostDataSource stream_id + else + return ()++ continue +++ Right (NH2.Frame (NH2.FrameHeader _ flags _) (NH2.PingFrame _)) | NH2.testAck flags-> do + -- Deal with pings: this is an Ack, so do nothing+ continue ++ Right (NH2.Frame (NH2.FrameHeader _ _ _) (NH2.PingFrame somebytes)) -> do + -- Deal with pings: NOT an Ack, so answer+ liftIO $ debugM "HTTP2.Session" "Ping processed"+ sendOutFrame+ (NH2.EncodeInfo+ (NH2.setAck NH2.defaultFlags)+ (NH2.toStreamIdentifier 0)+ Nothing + )+ (NH2.PingFrame somebytes)++ continue ++ Right (NH2.Frame frame_header (NH2.SettingsFrame _)) | isSettingsAck frame_header -> do + -- Frame was received by the peer, do nothing here...+ continue +++ Right (NH2.Frame _ (NH2.SettingsFrame settings_list)) -> do + liftIO $ debugM "HTTP2.Session" $ "Received settings: " ++ (show settings_list)+ -- Just acknowledge the frame.... for now + sendOutFrame + (NH2.EncodeInfo+ (NH2.setAck NH2.defaultFlags)+ (NH2.toStreamIdentifier 0)+ Nothing )+ (NH2.SettingsFrame [])++ continue +++ Right somethingelse -> do + liftIO $ errorM "HTTP2.Session" $ "Received problematic frame: "+ liftIO $ errorM "HTTP2.Session" $ ".. " ++ (show somethingelse)++ continue ++ where ++ continue = sessionInputThread++ sendOutFrame :: NH2.EncodeInfo -> NH2.FramePayload -> ReaderT SessionData IO ()+ sendOutFrame encode_info payload = do + session_output_mvar <- view sessionOutput + session_output <- liftIO $ takeMVar session_output_mvar++ liftIO $ writeChan session_output $ Right (encode_info, payload)++ liftIO $ putMVar session_output_mvar session_output+++frameEndsStream :: InputFrame -> Bool +frameEndsStream (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndStream flags+++createMechanismForStream :: GlobalStreamId -> ReaderT SessionData IO PostInputMechanism+createMechanismForStream stream_id = do + (chan, source) <- liftIO $ unfoldChannelAndSource+ stream2postinputmechanism <- view stream2PostInputMechanism+ let pim = PostInputMechanism (chan, source)+ liftIO $ H.insert stream2postinputmechanism stream_id pim + return pim+++-- TODO: Can be optimized by factoring out the mechanism lookup+closePostDataSource :: GlobalStreamId -> ReaderT SessionData IO ()+closePostDataSource stream_id = do + stream2postinputmechanism <- view stream2PostInputMechanism++ pim_maybe <- liftIO $ H.lookup stream2postinputmechanism stream_id ++ case pim_maybe of ++ Just (PostInputMechanism (chan, _)) -> + liftIO $ writeChan chan Nothing++ Nothing -> + -- This is an internal error, the mechanism should be + -- created when the stream ends+ error "Internal error/closePostDataSource"+++streamWorkerSendData :: Int -> B.ByteString -> ReaderT SessionData IO ()+streamWorkerSendData stream_id bytes = do + s2pim <- view stream2PostInputMechanism+ pim_maybe <- liftIO $ H.lookup s2pim stream_id ++ case pim_maybe of ++ Just pim -> + sendBytesToPim pim bytes++ Nothing -> + -- This is an internal error, the mechanism should be + -- created when the headers end+ error "Internal error"+++sendBytesToPim :: PostInputMechanism -> B.ByteString -> ReaderT SessionData IO ()+sendBytesToPim (PostInputMechanism (chan, _)) bytes = + liftIO $ writeChan chan (Just bytes)+++postDataSourceFromMechanism :: PostInputMechanism -> InputDataStream+postDataSourceFromMechanism (PostInputMechanism (_, source)) = source+++isSettingsAck :: NH2.FrameHeader -> Bool +isSettingsAck (NH2.FrameHeader _ flags _) = + NH2.testAck flags+++isStreamCancelled :: GlobalStreamId -> WorkerMonad Bool +isStreamCancelled stream_id = do + cancelled_streams_mvar <- view streamsCancelled_WTE+ cancelled_streams <- liftIO $ readMVar cancelled_streams_mvar+ return $ NS.member stream_id cancelled_streams+++workerThread :: Request -> CoherentWorker -> WorkerMonad ()+workerThread req coherent_worker =+ do+ headers_output <- view headersOutput+ stream_id <- view streamId+ (headers, _, data_and_conclussion) <- liftIO $ coherent_worker req++ -- liftIO $ putStrLn $ "Num pushed streams: " ++ (show $ length pushed_streams)++ -- Now I send the headers, if that's possible at all+ headers_sent <- liftIO $ newEmptyMVar+ liftIO $ writeChan headers_output (stream_id, headers_sent, headers)++ -- At this moment I should ask if the stream hasn't been cancelled by the browser before+ -- commiting to the work of sending addtitional data+ is_stream_cancelled <- isStreamCancelled stream_id+ if not is_stream_cancelled++ then do+ -- I have a beautiful source that I can de-construct...+ -- TODO: Optionally pulling data out from a Conduit ....+ -- liftIO ( data_and_conclussion $$ (_sendDataOfStream stream_id) )+ -- + -- This threadlet should block here waiting for the headers to finish going+ (maybe_footers, _) <- runConduit $+ (transPipe liftIO data_and_conclussion) + `fuseBothMaybe` + (sendDataOfStream stream_id headers_sent)+ -- BIG TODO: Send the headers ... likely stream conclusion semantics + -- will need to be changed. + return ()+ else ++ return ()++-- v-- comp. monad.+sendDataOfStream :: GlobalStreamId -> MVar HeadersSent -> Sink B.ByteString (ReaderT WorkerThreadEnvironment IO) ()+sendDataOfStream stream_id headers_sent = do+ data_output <- view dataOutput+ transPipe liftIO $ do + -- Wait for permission to send the data+ liftIO $ takeMVar headers_sent+ foldMapM $ \ bytes ->+ writeChan data_output (stream_id, bytes)+++-- sendDataOfStream :: Sink +++-- Allow this very important function to be used in the future to process footers++-- difficultFunction :: (Monad m)+-- => ConduitM () a2 m r1 -> ConduitM a2 Void m r2+-- -> m (r2, Maybe r1)+-- difficultFunction l r = liftM (fmap getLast) $ runWriterT (l' $$ r')+-- where+-- l' = transPipe lift l >>= lift . tell . Last . Just+-- r' = transPipe lift r++++appendHeaderFragmentBlock :: GlobalStreamId -> B.ByteString -> ReaderT SessionData IO ()+appendHeaderFragmentBlock global_stream_id bytes = do + ht <- view stream2HeaderBlockFragment + maybe_old_block <- liftIO $ H.lookup ht global_stream_id+ new_block <- return $ case maybe_old_block of ++ Nothing -> bytes++ Just something -> something `B.append` bytes ++ liftIO $ H.insert ht global_stream_id new_block+++getHeaderBytes :: GlobalStreamId -> ReaderT SessionData IO B.ByteString+getHeaderBytes global_stream_id = do + ht <- view stream2HeaderBlockFragment + Just bytes <- liftIO $ H.lookup ht global_stream_id+ return bytes+++frameIsHeaderOfStream :: InputFrame -> Maybe (GlobalStreamId, B.ByteString)+frameIsHeaderOfStream (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.HeadersFrame _ block_fragment ) )+ = Just (NH2.fromStreamIdentifier stream_id, block_fragment)+frameIsHeaderOfStream (NH2.Frame (NH2.FrameHeader _ _ stream_id) ( NH2.ContinuationFrame block_fragment) )+ = Just (NH2.fromStreamIdentifier stream_id, block_fragment)+frameIsHeaderOfStream _ + = Nothing +++frameEndsHeaders :: InputFrame -> Bool +frameEndsHeaders (NH2.Frame (NH2.FrameHeader _ flags _) _) = NH2.testEndHeader flags+++streamIdFromFrame :: InputFrame -> GlobalStreamId+streamIdFromFrame (NH2.Frame (NH2.FrameHeader _ _ stream_id) _) = NH2.fromStreamIdentifier stream_id+++-- TODO: Have different size for the headers..... just now going with a default size of 16 k...+-- TODO: Find a way to kill this thread....+headersOutputThread :: Chan (GlobalStreamId, MVar HeadersSent, Headers)+ -> MVar (Chan (Either SessionOutputCommand OutputFrame)) + -> ReaderT SessionData IO ()+headersOutputThread input_chan session_output_mvar = forever $ do + (stream_id, headers_ready_mvar, headers) <- liftIO $ readChan input_chan+ -- liftIO $ putStrLn $ "Output headers: " ++ (show headers)++ -- First encode the headers using the table+ encode_dyn_table_mvar <- view toEncodeHeaders++ encode_dyn_table <- liftIO $ takeMVar encode_dyn_table_mvar+ (new_dyn_table, data_to_send ) <- liftIO $ HP.encodeHeader HP.defaultEncodeStrategy encode_dyn_table headers+ liftIO $ putMVar encode_dyn_table_mvar new_dyn_table++ -- Now split the bytestring in chunks of the needed size.... + let bs_chunks = bytestringChunk useChunkLength data_to_send++ -- And send the chunks through while locking the output place....+ session_output <- liftIO $ takeMVar session_output_mvar++ -- First frame is just a headers frame....+ if (length bs_chunks) == 1 + then+ do + let flags = NH2.setEndHeader NH2.defaultFlags++ -- Write the first frame + liftIO $ writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = flags+ ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, ++ NH2.HeadersFrame Nothing (head bs_chunks)+ )+ else + do + let flags = NH2.defaultFlags+ -- Write the first frame + liftIO $ writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = flags+ ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, ++ NH2.HeadersFrame Nothing (head bs_chunks)+ )+ -- And write the other frames+ let + writeContinuations :: [B.ByteString] -> ReaderT SessionData IO ()+ writeContinuations (last_fragment:[]) = liftIO $+ writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = NH2.setEndHeader NH2.defaultFlags + ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, ++ NH2.ContinuationFrame last_fragment+ )+ writeContinuations (fragment:xs) = do + liftIO $ writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = NH2.defaultFlags + ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, ++ NH2.ContinuationFrame fragment+ )+ writeContinuations xs+++ writeContinuations (tail bs_chunks)+ -- Restore output capability, so that other pieces waiting can send...+ liftIO $ putMVar session_output_mvar session_output+ -- And say that the headers for this thread are out + liftIO $ putMVar headers_ready_mvar HeadersSent+ -- liftIO $ putStrLn "Headers were output"+ ++bytestringChunk :: Int -> B.ByteString -> [B.ByteString]+bytestringChunk len s | (B.length s) < len = [ s ]+bytestringChunk len s = h:(bytestringChunk len xs)+ where + (h, xs) = B.splitAt len s +++-- TODO: find a clean way to finish this thread (maybe with negative stream ids?)+-- TODO: This function does non-optimal chunking for the case where responses are+-- actually streamed.... in those cases we need to keep state for frames in +-- some other format.... +dataOutputThread :: Chan (GlobalStreamId, B.ByteString)+ -> MVar (Chan (Either SessionOutputCommand OutputFrame)) + -> IO ()+dataOutputThread input_chan session_output_mvar = forever $ do + (stream_id, contents) <- readChan input_chan++ -- And now just simply output it...+ let bs_chunks = bytestringChunk useChunkLength contents+ -- putStrLn $ "Chunk lengths: " ++ (show (map B.length bs_chunks))++ -- And send the chunks through while locking the output place....+ session_output <- liftIO $ takeMVar session_output_mvar++ -- First frame is the only one:+ if (length bs_chunks) == 1 + then+ do + let flags = NH2.setEndStream NH2.defaultFlags++ -- Write the first frame + liftIO $ writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = flags+ ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, ++ NH2.DataFrame (head bs_chunks)+ )+ else + do + let flags = NH2.defaultFlags+ -- Write the first frame + liftIO $ writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = flags+ ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, ++ NH2.DataFrame (head bs_chunks)+ )+ -- And write the other frames+ let + writeContinuations :: [B.ByteString] -> IO ()+ writeContinuations (last_fragment:[]) = liftIO $+ writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = NH2.setEndStream NH2.defaultFlags + ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, ++ NH2.DataFrame last_fragment+ )+ writeContinuations (fragment:xs) = do + liftIO $ writeChan session_output $ Right ( NH2.EncodeInfo {+ NH2.encodeFlags = NH2.defaultFlags + ,NH2.encodeStreamId = NH2.toStreamIdentifier stream_id + ,NH2.encodePadding = Nothing }, ++ NH2.DataFrame fragment+ )+ writeContinuations xs+ writeContinuations (tail bs_chunks)+ -- Restore output capability, so that other pieces waiting can send...+ liftIO $ debugM "HTTP2.Session" $ "Output capability restored"+ liftIO $ putMVar session_output_mvar session_output
+ hs-src/SecondTransfer/MainLoop.hs view
@@ -0,0 +1,26 @@+module SecondTransfer.MainLoop (+ -- * Callback types+ Attendant+ ,PullAction+ ,PushAction+ ,CloseAction+ ,IOProblem+ ,GenericIOProblem+ -- * High level OpenSSL functions. + -- + -- | Use these functions to create your TLS-compliant + -- HTTP/2 server in a snap.+ ,tlsServeWithALPN+ ,tlsServeWithALPNAndFinishOnRequest++ ,TLSLayerGenericProblem(..)+ ,FinishRequest(..)+ ) where +++import SecondTransfer.MainLoop.PushPullType (Attendant, PullAction,+ PushAction, CloseAction,+ IOProblem, GenericIOProblem+ )++import SecondTransfer.MainLoop.OpenSSL_TLS
+ hs-src/SecondTransfer/MainLoop/CoherentWorker.hs view
@@ -0,0 +1,101 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FunctionalDependencies, FlexibleInstances, DeriveDataTypeable #-} +-- | A CoherentWorker is one that doesn't need to compute everything at once...+-- This one is simpler than the SPDY one, because it enforces certain order....++++module SecondTransfer.MainLoop.CoherentWorker(+ getHeaderFromFlatList++ , Headers+ , FinalizationHeaders+ , Request+ , Footers+ , CoherentWorker+ , PrincipalStream+ , PushedStreams+ , PushedStream+ , DataAndConclusion+ , InputDataStream+ , StreamCancelledException (..)+ ) where ++import Control.Exception+import qualified Data.ByteString as B+import Data.Conduit+import Data.Foldable (find)+import Data.Typeable+++-- |List of headers. The first part of each tuple is the header name +-- (be sure to conform to the HTTP/2 convention of using lowercase)+-- and the second part is the headers contents. This list needs to include+-- the special :method, :scheme, :authority and :path pseudo-headers for +-- requests; and :status (with a plain numeric value represented in ascii digits)+-- for responses.+type Headers = [(B.ByteString, B.ByteString)]++-- |This is a Source conduit (see Haskell Data.Conduit library from Michael Snoyman)+-- that you can use to retrieve the data sent by the client piece-wise. +type InputDataStream = Source IO B.ByteString++-- | A request is a set of headers and a request body....+-- which will normally be empty, except for POST and PUT requests. But +-- this library enforces none of that. +type Request = (Headers, Maybe InputDataStream)++-- | Finalization headers. If you don't know what they are, chances are +-- that you don't need to worry about them for now. The support in this +-- library for those are at best sketchy. +type FinalizationHeaders = Headers++-- | Finalization headers +type Footers = FinalizationHeaders++-- | You use this type to answer a request. The `Headers` are thus response +-- headers and they should contain the :status pseudo-header. The `PushedStreams`+-- is a list of pushed streams...(I don't thaink that I'm handling those yet)+type PrincipalStream = (Headers, PushedStreams, DataAndConclusion)+++-- | A source-like conduit with the data returned in the response. The +-- return value of the conduit is a list of footers. For now that list can +-- be anything (even bottom), I'm not handling it just yet. +type DataAndConclusion = ConduitM () B.ByteString IO Footers++-- | Main type of this library. You implement one of these for your server.+-- Basically this is a callback that the library calls as soon as it has+-- all the headers of a request. For GET requests that's the entire request+-- basically, but for POST and PUT requests this is just before the data +-- starts arriving to the server. +type CoherentWorker = Request -> IO PrincipalStream++-- | This exception will be raised inside a `CoherentWorker` when the underlying +-- stream is cancelled (STREAM_RESET in HTTP\/2). Do any necessary cleanup+-- in a handler, or simply use the fact that the exception is asynchronously+-- delivered +-- to your CoherentWorker Haskell thread, giving you an opportunity to +-- interrupt any blocked operations.+data StreamCancelledException = StreamCancelledException+ deriving (Show, Typeable)++instance Exception StreamCancelledException++-- | A list of pushed streams +type PushedStreams = [ IO PushedStream ]++-- | A pushed stream, represented by a list of request headers, +-- a list of response headers, and the usual response body (which +-- may include final footers (not implemented yet)).+type PushedStream = (Headers, Headers, DataAndConclusion)++-- | Gets a single header from the list+getHeaderFromFlatList :: Headers -> B.ByteString -> Maybe B.ByteString+getHeaderFromFlatList unvl bs = + case find (\ (x,_) -> x==bs ) unvl of+ Just (_, found_value) -> Just found_value ++ Nothing -> Nothing +
+ hs-src/SecondTransfer/MainLoop/Framer.hs view
@@ -0,0 +1,78 @@+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.MainLoop.Framer(+ readNextChunk+ ,readLength++ ,Framer+ ,LengthCallback+ ) where+++import Control.Monad.Trans.Class (lift)+import Control.Monad.IO.Class (MonadIO+ -- , liftIO+ )+import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as LB+import Data.Conduit++import Data.Monoid (mappend, mempty)+++type Framer m = LB.ByteString -- Input left overs+ -> m B.ByteString -- Generator+ -> Maybe Int -- Length to read, if we know now+ -> m (LB.ByteString, LB.ByteString) -- To yield, left-overs...++++-- * Doing it by parts++type LengthCallback = B.ByteString -> Maybe Int+++readNextChunk :: Monad m =>+ LengthCallback -- ^ How to know if we can split somewhere+ -> B.ByteString -- ^ Input left-overs+ -> m B.ByteString -- ^ Generator action+ -> Source m B.ByteString -- ^ Packet and leftovers, if we could get them +readNextChunk length_callback input_leftovers gen = do + let + maybe_length = length_callback input_leftovers+ readUpTo lo the_length | (B.length lo) >= the_length = + return $ B.splitAt the_length lo+ readUpTo lo the_length = do + frag <- lift gen + readUpTo (lo `mappend` frag) the_length++ case maybe_length of + Just the_length -> do + -- Just need to read the rest .... + (package_bytes, newnewleftovers) <- readUpTo input_leftovers the_length+ yield package_bytes + readNextChunk length_callback newnewleftovers gen ++ Nothing -> do + -- Read a bit more + new_fragment <- lift gen + let new_leftovers = input_leftovers `mappend` new_fragment+ readNextChunk length_callback new_leftovers gen+++-- Some protocols, e.g., http/2, have the client transmit a fixed-length+-- prefix. This function reads both that prefix and returns whatever get's+-- trapped up there.... +readLength :: MonadIO m => Int -> m B.ByteString -> m (B.ByteString, B.ByteString)+readLength the_length gen = + readUpTo mempty + where + readUpTo lo + | (B.length lo) >= the_length = do+ -- liftIO $ putStrLn "Full read"+ return $ B.splitAt the_length lo+ | otherwise = do + -- liftIO $ putStrLn $ "fragment read " ++ (show lo) + frag <- gen + readUpTo (lo `mappend` frag)++
+ hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs view
@@ -0,0 +1,350 @@+{-# LANGUAGE ForeignFunctionInterface, OverloadedStrings, DeriveDataTypeable #-}+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.MainLoop.OpenSSL_TLS(+ tlsServeWithALPN+ ,tlsServeWithALPNAndFinishOnRequest+ -- ,tlsServeWithALPNOnce++ ,TLSLayerGenericProblem(..)+ ,FinishRequest(..)+ ) where ++++import Control.Monad+import Control.Concurrent.MVar +import Control.Exception +import qualified Control.Exception as E+import Data.Foldable (foldMap)+import Data.Typeable +import Data.Monoid ()+import Foreign+import Foreign.C++import qualified Data.ByteString as B+import qualified Data.ByteString.Builder as BB+import Data.ByteString.Char8 (pack)+import qualified Data.ByteString.Lazy as LB+import qualified Data.ByteString.Unsafe as BU++import System.Log.Logger++import SecondTransfer.MainLoop.PushPullType+ +++-- | Exception inheriting from `IOProblem`. This is thrown by the +-- OpenSSL subsystem to signal that the connection was broken or that +-- otherwise there was a problem at the SSL layer. +data TLSLayerGenericProblem = TLSLayerGenericProblem String+ deriving (Show, Typeable)+++instance Exception TLSLayerGenericProblem where + toException = toException . IOProblem + fromException x = do + IOProblem a <- fromException x + cast a+++data InterruptibleEither a b = + Left_I a + |Right_I b + |Interrupted+++-- | Singleton type. Used in conjunction with an `MVar`. If the MVar is full, +-- the fuction `tlsServeWithALPNAndFinishOnRequest` knows that it should finish+-- at its earliest convenience and call the `CloseAction` for any open sessions.+data FinishRequest = FinishRequest+++-- These names are absolutely improper....+-- Session creator+data Connection_t +-- Session+data Wired_t++type Connection_Ptr = Ptr Connection_t +type Wired_Ptr = Ptr Wired_t+++-- Actually, this makes a listener for new connections+-- connection_t* make_connection(char* certificate_filename, char* privkey_filename, char* hostname, int portno, +-- char* protocol_list, int protocol_list_len)+foreign import ccall "make_connection" makeConnection :: + CString -- cert filename+ -> CString -- privkey_filename+ -> CString -- hostname+ -> CInt -- port+ -> Ptr CChar -- protocol list+ -> CInt -- protocol list length+ -> IO Connection_Ptr++allOk :: CInt +allOk = 0 ++badHappened :: CInt +badHappened = 1 ++timeoutReached :: CInt +timeoutReached = 3++-- int wait_for_connection(connection_t* conn, wired_session_t** wired_session);+foreign import ccall "wait_for_connection" waitForConnection :: Connection_Ptr -> CInt -> Ptr Wired_Ptr -> IO CInt ++-- int send_data(wired_session_t* ws, char* buffer, int buffer_size);+foreign import ccall "send_data" sendData :: Wired_Ptr -> Ptr CChar -> CInt -> IO CInt ++-- int recv_data(wired_session_t* ws, char* inbuffer, int buffer_size, int* data_recvd);+foreign import ccall "recv_data" recvData :: Wired_Ptr -> Ptr CChar -> CInt -> Ptr CInt -> IO CInt++-- int get_selected_protocol(wired_session_t* ws){ return ws->protocol_index; }+foreign import ccall "get_selected_protocol" getSelectedProtocol :: Wired_Ptr -> IO CInt++-- void dispose_wired_session(wired_session_t* ws);+foreign import ccall "dispose_wired_session" disposeWiredSession :: Wired_Ptr -> IO ()++foreign import ccall "close_connection" closeConnection :: Connection_Ptr -> IO ()+++useBufferSize :: Int+useBufferSize = 4096+++type Protocols = [B.ByteString]+++protocolsToWire :: Protocols -> B.ByteString+protocolsToWire protocols = + LB.toStrict . BB.toLazyByteString $ + foldMap (\ protocol + -> (BB.lazyByteString . LB.fromChunks)+ [ B.singleton $ fromIntegral $ B.length protocol,+ protocol + ]+ ) protocols+++-- | Simple function to open +tlsServeWithALPN :: FilePath -- ^ Path to a certificate the server is going to use to identify itself.+ -- Bear in mind that multiple domains can be served from the same HTTP/2 + -- TLS socket, so please create the HTTP/2 certificate accordingly.+ -> FilePath -- ^ Path to the key of your certificate. + -> String -- ^ Name of the network interface where you want to start your server+ -> [(String, Attendant)] -- ^ List of protocol names and the corresponding `Attendant` to use for + -- each. This way you can serve both HTTP\/1.1 over TLS and HTTP\/2 in the+ -- same socket.+ -> Int -- ^ Port to open to listen for connections. + -> IO ()+tlsServeWithALPN certificate_filename key_filename interface_name attendants interface_port = do ++ let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants+ withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do ++ connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) ->+ makeConnection + c_certfn+ c_keyfn+ c_iname+ (fromIntegral interface_port)+ pchar + (fromIntegral len)++ if connection_ptr == nullPtr + then + throwIO $ TLSLayerGenericProblem "Could not create listening end"+ else + return ()++ forever $ do + either_wired_ptr <- alloca $ \ wired_ptr_ptr -> + let + tryOnce = do + result_code <- waitForConnection connection_ptr defaultWaitTime wired_ptr_ptr+ let + r = case result_code of + re | re == allOk -> do + p <- peek wired_ptr_ptr+ return $ Right p+ | re == timeoutReached -> tryOnce + | re == badHappened -> return $ Left "A wait for connection failed"+ r + in tryOnce++ case either_wired_ptr of ++ Left msg -> do + errorM "OpenSSL" $ ".. wait for connection failed. " ++ msg+++ Right wired_ptr -> do + already_closed_mvar <- newMVar False+ let + pushAction datum = BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do + result <- sendData wired_ptr pchar (fromIntegral len)+ case result of + r | r == allOk -> return ()+ | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not send data"+ pullAction = do + allocaBytes useBufferSize $ \ pcharbuffer -> + alloca $ \ data_recvd_ptr -> do + result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr+ recvd_bytes <- case result of + r | r == allOk -> peek data_recvd_ptr+ | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not receive data"++ B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes)++ -- Ensure that the socket and the struct are only closed once+ closeAction = do+ b <- readMVar already_closed_mvar+ if not b + then do+ putMVar already_closed_mvar True+ disposeWiredSession wired_ptr+ else + return ()++ use_protocol <- getSelectedProtocol wired_ptr++ -- putStrLn $ ".. Using protocol: " ++ (show use_protocol)++ let + maybe_session_attendant = case fromIntegral use_protocol of + n | (use_protocol >= 0) -> Just $ snd $ attendants !! n + | otherwise -> Nothing ++ case maybe_session_attendant of ++ Just session_attendant -> + E.catch + (session_attendant pushAction pullAction closeAction)+ ((\ e -> do + errorM "OpenSSL" " ** Session ended by TLSLayerGenericProblem (well handled)"+ throwIO e+ )::TLSLayerGenericProblem -> IO () )+++ Nothing ->+ return ()+++-- | Interruptible version of `tlsServeWithALPN`. Use the extra argument to notify +-- the server of finishing. +tlsServeWithALPNAndFinishOnRequest :: FilePath + -> FilePath -- ^ Same as for `tlsServeWithALPN` + -> String -- ^ Same as for `tlsServeWithALPN`+ -> [(String, Attendant)] -- ^ Same as for `tlsServeWithALPN`+ -> Int -- ^ Same as for `tlsServeWithALPN`+ -> MVar FinishRequest -- ^ Finish request+ -> IO ()+tlsServeWithALPNAndFinishOnRequest certificate_filename key_filename interface_name attendants interface_port finish_request = do ++ let protocols_bs = protocolsToWire $ fmap (\ (s,_) -> pack s) attendants+ withCString certificate_filename $ \ c_certfn -> withCString key_filename $ \ c_keyfn -> withCString interface_name $ \ c_iname -> do ++ -- Create an accepting endpoint+ connection_ptr <- BU.unsafeUseAsCStringLen protocols_bs $ \ (pchar, len) ->+ makeConnection + c_certfn+ c_keyfn+ c_iname+ (fromIntegral interface_port)+ pchar + (fromIntegral len)++ -- Create a computation that accepts a connection, runs a session on it and recurses+ let + recursion = do + -- Get a SSL session+ either_wired_ptr <- alloca $ \ wired_ptr_ptr -> + let + tryOnce = do + result_code <- waitForConnection connection_ptr smallWaitTime wired_ptr_ptr+ let + r = case result_code of + re | re == allOk -> do + p <- peek wired_ptr_ptr+ return $ Right_I p+ | re == timeoutReached -> do + got_finish_request <- tryTakeMVar finish_request+ case got_finish_request of + Nothing ->+ tryOnce+ Just _ ->+ return Interrupted ++ | re == badHappened -> return $ Left_I "A wait for connection failed"+ r + in tryOnce++ -- With the potentially obtained SSL session do...+ case either_wired_ptr of ++ Left_I msg -> do + errorM "OpenSSL" $ ".. wait for connection failed. " ++ msg++ -- // .. //+ recursion++ Right_I wired_ptr -> do + already_closed_mvar <- newMVar False+ let + pushAction datum = BU.unsafeUseAsCStringLen (LB.toStrict datum) $ \ (pchar, len) -> do + result <- sendData wired_ptr pchar (fromIntegral len)+ case result of + r | r == allOk -> return ()+ | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not send data"+ pullAction = do + allocaBytes useBufferSize $ \ pcharbuffer -> + alloca $ \ data_recvd_ptr -> do + result <- recvData wired_ptr pcharbuffer (fromIntegral useBufferSize) data_recvd_ptr+ recvd_bytes <- case result of + r | r == allOk -> peek data_recvd_ptr+ | r == badHappened -> throwIO $ TLSLayerGenericProblem "Could not receive data"++ B.packCStringLen (pcharbuffer, fromIntegral recvd_bytes)+ closeAction = do+ b <- readMVar already_closed_mvar+ if not b + then do+ putMVar already_closed_mvar True+ disposeWiredSession wired_ptr+ else + return ()++ use_protocol <- getSelectedProtocol wired_ptr++ infoM "OpenSSL" $ ".. Using protocol: " ++ (show use_protocol)++ let + maybe_session_attendant = case fromIntegral use_protocol of + n | (use_protocol >= 0) -> Just $ snd $ attendants !! n + | otherwise -> Nothing ++ case maybe_session_attendant of ++ Just session_attendant -> + session_attendant pushAction pullAction closeAction++ Nothing ->+ return ()++ -- // .. //+ recursion ++ Interrupted -> do+ infoM "OpenSSL" "Connection closed"+ closeConnection connection_ptr++ -- Start the loop defined above...+ recursion ++-- When we are using the eternal version of this function, wake up +-- each second .... +defaultWaitTime :: CInt+defaultWaitTime = 200000+-- Okej, more responsiviness needed +smallWaitTime :: CInt +smallWaitTime = 50000
+ hs-src/SecondTransfer/MainLoop/PushPullType.hs view
@@ -0,0 +1,71 @@+{-# LANGUAGE DeriveDataTypeable, ExistentialQuantification #-}+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.MainLoop.PushPullType (+ PushAction+ ,PullAction+ ,Attendant+ ,CloseAction+ ,IOProblem(..)+ ,GenericIOProblem(..)+ ) where +++import Control.Exception +import Data.Typeable (Typeable, cast)++import qualified Data.ByteString as B+import qualified Data.ByteString.Lazy as LB++-- | Callback type to push data to a channel. Part of this +-- interface is the abstract exception type IOProblem. Throw an +-- instance of it to notify the session that the connection has +-- been broken. +type PushAction = LB.ByteString -> IO ()++-- | Callback type to pull data from a channel. The same+-- as to PushAction applies to exceptions thrown from +-- there. +type PullAction = IO B.ByteString++-- | Callback that the Session calls to realease resources +-- associated with the channels. Take into account that your+-- callback should be able to deal with non-clean shutdowns+-- also, for example, if the connection to the remote peer+-- is severed suddenly.+type CloseAction = IO ()++-- | A function which takes three arguments: the first one says +-- how to send data (on a socket or similar transport), and the second one how +-- to receive data on said socket. The third argument encapsulates +-- the sequence of steps needed for a clean shutdown. +--+-- You can implement one of these to let somebody else supply the +-- push, pull and close callbacks. In this library we supply callbacks+-- for TLS sockets, so that you don't need to go through the drudgery +-- of managing those yourself.+type Attendant = PushAction -> PullAction -> CloseAction -> IO () +++-- | Throw exceptions derived from this (e.g, `GenericIOProblem` below)+-- to have the HTTP/2 session to terminate gracefully. +data IOProblem = forall e . Exception e => IOProblem e + deriving Typeable+++instance Show IOProblem where+ show (IOProblem e) = show e ++instance Exception IOProblem ++-- | A concrete case of the above exception. Throw one of this+-- if you don't want to implement your own type. Capture one +-- `IOProblem` otherwise. +data GenericIOProblem = GenericIOProblem+ deriving (Show, Typeable)+++instance Exception GenericIOProblem where + toException = toException . IOProblem+ fromException x = do + IOProblem a <- fromException x + cast a
+ hs-src/SecondTransfer/MainLoop/Tokens.hs view
@@ -0,0 +1,150 @@+{-# LANGUAGE ScopedTypeVariables #-}+{-# LANGUAGE OverloadedStrings #-}+{-# LANGUAGE FunctionalDependencies, FlexibleInstances #-} +{-# OPTIONS_HADDOCK hide #-}++module SecondTransfer.MainLoop.Tokens(+ packHeaderTuples+ ,unpackHeaderTuples+ ,getHeader+ ,actionIsForAssociatedStream++ ,UnpackedNameValueList (..)+ ,StreamInputToken (..)+ ,StreamOutputAction (..)+ ,StreamWorker+ ,StreamWorkerClass (..)+ ,LocalStreamId+ ,GlobalStreamId+ ) where ++++import Control.Monad (forM_, replicateM)+import Data.Binary (Binary, get, put)+import Data.Binary.Get (getByteString, getWord32be)+import Data.Binary.Put (putWord32be, putByteString)+import qualified Data.ByteString as B+import qualified Data.ByteString as BS+import Data.Conduit (Conduit)+import Data.List (sortBy, find)+import Data.Word++++-- Not to use raw....+newtype UnpackedNameValueList = UnpackedNameValueList [(B.ByteString, B.ByteString)]+ deriving Show+++++data StreamInputToken = Headers_STk UnpackedNameValueList+ | Data_Stk B.ByteString+ | Finish_Stk+ deriving Show+++type LocalStreamId = Int+++type GlobalStreamId = Int+++data StreamOutputAction = SendHeaders_SOA UnpackedNameValueList + | SendAssociatedHeaders_SOA LocalStreamId UnpackedNameValueList+ | SendData_SOA B.ByteString + | SendAssociatedData_SOA LocalStreamId B.ByteString+ | SendAssociatedFinish_SOA LocalStreamId+ | Finish_SOA+ deriving Show+++actionIsForAssociatedStream :: StreamOutputAction -> Maybe (LocalStreamId, StreamOutputAction)+actionIsForAssociatedStream (SendAssociatedData_SOA stream_id x ) = Just (stream_id, SendData_SOA x)+actionIsForAssociatedStream (SendAssociatedHeaders_SOA stream_id x ) = Just (stream_id, SendHeaders_SOA x)+actionIsForAssociatedStream (SendAssociatedFinish_SOA stream_id ) = Just (stream_id, Finish_SOA )+actionIsForAssociatedStream _ = Nothing +++-- | A StreamWorker: a conduit that takes input tokens and answers with output +-- tokens. It can perform I/O.+type StreamWorker = Conduit StreamInputToken IO StreamOutputAction+++-- | Sequence of steps to get a StreamWorker. This class is independent of things+-- like the finer details concerning the frames and the streams.+--+-- Todo: although this shows a common pattern, I'm not sure how having a class+-- here helps....+class StreamWorkerClass serviceParams servicePocket sessionPocket | + serviceParams -> sessionPocket servicePocket,+ servicePocket -> sessionPocket serviceParams,+ sessionPocket -> servicePocket where++ initService :: serviceParams -> IO servicePocket++ initSession :: servicePocket -> IO sessionPocket++ initStream :: servicePocket -> sessionPocket -> IO StreamWorker++++ +instance Binary UnpackedNameValueList where + put unvl = + do + putWord32be length32+ forM_ packed $ \ (h,v) -> do + putWord32be $ fromIntegral (BS.length h)+ putByteString h++ putWord32be $ fromIntegral (BS.length v)+ putByteString v+ where + length32 = (fromIntegral $ length packed)::Word32 + packed = packHeaderTuples unvl++ get = + do+ entry_count <- getWord32be+ packed_entries <- replicateM (fromIntegral entry_count) $ do { + name_length <- getWord32be+ ; name <- getByteString (fromIntegral name_length)+ ; value_length <- getWord32be + ; value <- getByteString (fromIntegral value_length)+ ; return (name, value) }+ return $ unpackHeaderTuples packed_entries +++-- Just puts them together, as per the spec+packHeaderTuples :: UnpackedNameValueList -> [(BS.ByteString, BS.ByteString)]+packHeaderTuples (UnpackedNameValueList uvl) = let+ sortFun (h1, _) (h2, _) = compare h1 h2+ sorted_uvl = sortBy sortFun uvl + sameName [] = []+ sameName ((h, v):rest) = let + (cousins,nocousins) = span (\ (hh, _) -> hh == h ) rest+ cousings_value = BS.intercalate "\0" $ v:(map snd cousins)+ in + (h, cousings_value):(sameName nocousins)+ in + sameName sorted_uvl+++-- And unputs them together +unpackHeaderTuples :: [(BS.ByteString, BS.ByteString)] -> UnpackedNameValueList+unpackHeaderTuples [] = UnpackedNameValueList []+unpackHeaderTuples vl = UnpackedNameValueList $ step vl+ where + valueSplit v = BS.split 0 v+ step [] = []+ step ((h,v):rest) = [ (h,vv) | vv <- valueSplit v ] ++ (step rest) +++getHeader :: UnpackedNameValueList -> BS.ByteString -> Maybe BS.ByteString+getHeader (UnpackedNameValueList unvl) bs = + case find (\ (x,_) -> x==bs ) unvl of+ Just (_, found_value) -> Just found_value ++ Nothing -> Nothing
+ hs-src/SecondTransfer/Utils.hs view
@@ -0,0 +1,153 @@+{-# LANGUAGE OverloadedStrings, GeneralizedNewtypeDeriving #-}+{-# OPTIONS_HADDOCK hide #-}+module SecondTransfer.Utils (+ strToInt+ ,Word24+ ,word24ToInt+ ,putWord24be+ ,getWord24be+ -- ,getTimeDiff+ -- ,timeAsDouble + -- ,reportTimedEvent+ ,lowercaseText+ ,unfoldChannelAndSource+ ,stripString+ -- ,neutralizeUrl+ ,domainFromUrl+ -- ,hashFromUrl+ -- ,hashSafeFromUrl+ -- ,unSafeUrl++ -- ,SafeUrl+ ) where +++import Control.Concurrent.Chan+import Control.Monad.Trans.Class (lift)+-- import qualified Crypto.Hash.MD5 as MD5+import Data.Binary (Binary, get, put, putWord8)+import Data.Binary.Get (Get, getWord16be, getWord8)+import Data.Binary.Put (Put, putWord16be)+import Data.Bits+import qualified Data.ByteString as B+-- import qualified Data.ByteString.Base16 as B16+import Data.ByteString.Char8 (pack, unpack)+-- import Data.Hashable (Hashable)+import Data.Conduit+import qualified Data.Text as T+import Data.Text.Encoding+import qualified Network.URI as U+-- import qualified System.Clock as SC+-- import Text.Printf (printf)+-- import qualified Text.Show.ByteString as S(Show(..))++++strToInt::String -> Int +strToInt = fromIntegral . toInteger . (read::String->Integer)+++newtype Word24 = Word24 Int+ deriving (Show)+++-- Newtype to protect url usage+-- newtype SafeUrl = SafeUrl { unSafeUrl :: B.ByteString } deriving (Eq, Show, Hashable)++++word24ToInt :: Word24 -> Int +word24ToInt (Word24 w24) = w24+++instance Binary Word24 where++ put (Word24 w24) = + do + let + high_stuff = w24 `shiftR` 24 + low_stuff = w24 `mod` (1 `shiftL` 24) + putWord8 $ fromIntegral high_stuff+ putWord16be $ fromIntegral low_stuff ++ get = do+ high_stuff <- getWord8 + low_stuff <- getWord16be+ let + value = (fromIntegral low_stuff) + ( (fromIntegral high_stuff) `shiftL` 24 ) + return $ Word24 value+++getWord24be :: Get Int+getWord24be = do + w24 <- get+ return $ word24ToInt w24+++putWord24be :: Int -> Put +putWord24be x = put (Word24 x)+++lowercaseText :: B.ByteString -> B.ByteString+lowercaseText bs0 = + encodeUtf8 ts1 + where + ts1 = T.toLower ts0 + ts0 = decodeUtf8 bs0+++unfoldChannelAndSource :: IO (Chan (Maybe a), Source IO a)+unfoldChannelAndSource = do + chan <- newChan + let + source = do + e <- lift $ readChan chan+ case e of + Just ee -> do+ yield ee+ source ++ Nothing -> + return ()++ return (chan, source)+++stripString :: String -> String +stripString = filter $ \ ch -> (ch /= '\n') && ( ch /= ' ')+++-- neutralizeUrl :: B.ByteString -> B.ByteString+-- neutralizeUrl url = let +-- Just (U.URI {- scheme -} _ authority u_path u_query u_frag) = U.parseURI $ unpack url+-- Just (U.URIAuth _ use_host _) = authority+-- complete_url = U.URI {+-- U.uriScheme = "snu:"+-- ,U.uriAuthority = Just $ U.URIAuth {+-- U.uriUserInfo = ""+-- ,U.uriRegName = use_host +-- ,U.uriPort = ""+-- }+-- ,U.uriPath = u_path+-- ,U.uriQuery = u_query +-- ,U.uriFragment = u_frag +-- }+-- in +-- pack $ show complete_url+++domainFromUrl :: B.ByteString -> B.ByteString+domainFromUrl url = let + Just (U.URI {- scheme -} _ authority _ _ _) = U.parseURI $ unpack url+ Just (U.URIAuth _ use_host _) = authority+ in + pack use_host+++-- hashFromUrl :: B.ByteString -> B.ByteString +-- hashFromUrl url = +-- B.take 10 . B16.encode . MD5.finalize $ foldl MD5.update MD5.init $ [urlHashSalt, neutralizeUrl url]+++-- hashSafeFromUrl :: B.ByteString -> SafeUrl +-- hashSafeFromUrl = SafeUrl . hashFromUrl
+ second-transfer.cabal view
@@ -0,0 +1,159 @@+-- Initial spdy-ping.cabal generated by cabal init. For further +-- documentation-- see http://haskell.org/cabal/users-guide/++-- The name of the package.+name : second-transfer++-- The package version. See the Haskell package versioning policy (PVP) +-- for standards guiding when and how versions should be incremented.+-- http://www.haskell.org/haskellwiki/Package_versioning_policy+-- PVP summary: +-+------- breaking API changes+-- | | +----- non-breaking API additions+-- | | | +--- code changes with no API change+version : 0.1.0.0++synopsis : Second Transfer HTTP/2 web server++description : Second Transfer HTTP/2 web server++homepage : www.zunzun.se++license : BSD3++license-file: LICENSE++author : Alcides Viamontes Esquivel++maintainer : alcidesv@zunzun.se++copyright : Copyright 2015, Alcides Viamontes Esquivel ++category : Network++stability : experimental++bug-reports : https://github.com/alcidesv/second-transfer/issues++build-type : Simple++-- Extra files to be distributed with the package-- such as examples or a +-- README.+extra-source-files: README.md++-- Constraint on the version of Cabal needed to build this package.+cabal-version: >=1.10++source-repository head+ type: git+ location: git@github.com:alcidesv/second-transfer.git++source-repository this+ type: git+ location: git@github.com:alcidesv/second-transfer.git+ tag: 0.1.0.0+++library++ exposed-modules: SecondTransfer+ , SecondTransfer.MainLoop+ , SecondTransfer.MainLoop.OpenSSL_TLS+ , SecondTransfer.Http2++ other-modules: SecondTransfer.MainLoop.CoherentWorker+ , SecondTransfer.MainLoop.PushPullType+ , SecondTransfer.MainLoop.Tokens+ , SecondTransfer.MainLoop.Framer+ , SecondTransfer.Utils+ , SecondTransfer.Http2.Framer+ , SecondTransfer.Http2.MakeAttendant+ , SecondTransfer.Http2.Session++ -- LANGUAGE extensions used by modules in this package.+ -- other-extensions: + + -- Other library packages from which modules are imported.+ build-depends: base >=4.7 && < 4.8,+ -- optparse-applicative+ -- aeson >= 0.8+ -- connection >= 0.2 && < 0.3+ -- exceptions >= 0.6+ bytestring == 0.10.4.0,+ -- base64-bytestring >= 1.0+ base16-bytestring >= 0.1.1,+ -- data-default == 0.5.3+ -- tls >= 1.2 && < 1.3+ -- data-default-class == 0.0.1+ network == 2.6.0.2,+ -- x509 == 1.5.0.1+ -- certificate == 1.3.9+ -- x509-store == 1.5.0+ -- x509-system == 1.5.0+ text >= 1.2 && < 1.3,+ -- bitset == 1.4.8+ binary == 0.7.1.0,+ -- pipes == 4.1.4+ containers == 0.5.5.1,+ -- clock == 0.4.1.3+ -- network-simple == 0.4.0.2+ -- pipes-concurrency >= 2.0 && < 2.1+ -- unix == 2.7.0.1+ -- filepath == 1.3.0.2+ -- directory == 1.2.1.0+ -- asn1-encoding == 0.9.0+ -- crypto-pubkey >= 0.2.7 && <0.3+ -- asn1-types == 0.3.0+ -- crypto-random == 0.0.8+ -- cryptohash >= 0.11,+ -- streaming-commons >= 0.1 && < 0.2+ conduit >= 1.2.4,+ -- conduit-combinators >= 0.3.0 && < 0.4+ transformers >=0.3 && <= 0.5,+ -- mmorph == 1.0.4+ -- QuickCheck == 2.7.6+ network-uri >= 2.6 && < 2.7,+ hashtables >= 1.2 && < 1.3,+ -- dequeue == 0.1.5+ lens >= 4.7 && < 4.8,+ -- base64-bytestring >= 1.0+ http2 >= 0.7,+ ---- hedis == 0.6.5+ -- blaze-builder >= 0.3.3 && < 0.4+ -- process >= 1.2 && < 1.3+ -- void + hslogger >= 1.2.6,+ hashable >= 1.2+ -- -- asn-data+ + -- Directories containing source files.+ hs-source-dirs: hs-src+ + -- Base language which the package is written in.+ default-language: Haskell2010++ -- Very specific directory with the version of openssl + -- that I'm using. As of February 2015-- this one is not + -- commonly installed + include-dirs: /opt/openssl-1.0.2/include++ c-sources: cbits/tlsinc.c++ -- cc-options: -fPIC -pthread -g -O0++ -- cc-options: -g3 -O0++ ghc-options:++ extra-libraries: ssl crypto++ extra-lib-dirs: /opt/openssl-1.0.2/lib++Test-Suite compiling-ok+ type : exitcode-stdio-1.0+ main-is : compiling_ok.hs+ hs-source-dirs : tests/tests-hs-src/+ default-language: Haskell2010+ build-depends : base >=4.7 && < 4.8,+ second-transfer,+ conduit >= 1.2.4+ ghc-options : -threaded
+ tests/tests-hs-src/compiling_ok.hs view
@@ -0,0 +1,44 @@+{-# LANGUAGE OverloadedStrings #-}+import SecondTransfer(+ CoherentWorker+ , Footers+ , DataAndConclusion+ , tlsServeWithALPN+ , http2Attendant+ )++import Data.Conduit+++saysHello :: DataAndConclusion+saysHello = do + yield "Hello world!"+ -- No footers+ return []+++helloWorldWorker :: CoherentWorker+helloWorldWorker request = return (+ [+ (":status", "200")+ ],+ [], -- No pushed streams+ saysHello+ )+++-- For this program to work, it should be run from the top of +-- the developement directory.+main = do + tlsServeWithALPN+ "tests/support/servercert.pem" -- Server certificate+ "tests/support/privkey.pem" -- Certificate private key+ "127.0.0.1" -- On which interface to bind+ [+ ("h2-14", http2_attendant), -- Protocols present in the ALPN negotiation+ ("h2", http2_attendant) -- they may be slightly different, but for this + -- test it doesn't matter.+ ]+ 8000+ where + http2_attendant = http2Attendant helloWorldWorker