diff --git a/LICENSE b/LICENSE new file mode 100644 --- /dev/null +++ b/LICENSE @@ -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. diff --git a/README.md b/README.md new file mode 100644 --- /dev/null +++ b/README.md @@ -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 diff --git a/Setup.hs b/Setup.hs new file mode 100644 --- /dev/null +++ b/Setup.hs @@ -0,0 +1,2 @@ +import Distribution.Simple +main = defaultMain diff --git a/cbits/tlsinc.c b/cbits/tlsinc.c new file mode 100644 --- /dev/null +++ b/cbits/tlsinc.c @@ -0,0 +1,785 @@ + +#define _BSD_SOURCE 1 +#define _XOPEN_SOURCE 1 /* or any value < 500 */ +#define _POSIX_C_SOURCE 1 + + + +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include +#include + +#include +#include +#include + +// 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 +#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; +} + diff --git a/hs-src/SecondTransfer.hs b/hs-src/SecondTransfer.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer.hs @@ -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 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 . 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) diff --git a/hs-src/SecondTransfer/Http2.hs b/hs-src/SecondTransfer/Http2.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Http2.hs @@ -0,0 +1,5 @@ +module SecondTransfer.Http2( + http2Attendant + ) where + +import SecondTransfer.Http2.MakeAttendant(http2Attendant) diff --git a/hs-src/SecondTransfer/Http2/Framer.hs b/hs-src/SecondTransfer/Http2/Framer.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Http2/Framer.hs @@ -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 diff --git a/hs-src/SecondTransfer/Http2/MakeAttendant.hs b/hs-src/SecondTransfer/Http2/MakeAttendant.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Http2/MakeAttendant.hs @@ -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 diff --git a/hs-src/SecondTransfer/Http2/Session.hs b/hs-src/SecondTransfer/Http2/Session.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Http2/Session.hs @@ -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 diff --git a/hs-src/SecondTransfer/MainLoop.hs b/hs-src/SecondTransfer/MainLoop.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/MainLoop.hs @@ -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 diff --git a/hs-src/SecondTransfer/MainLoop/CoherentWorker.hs b/hs-src/SecondTransfer/MainLoop/CoherentWorker.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/MainLoop/CoherentWorker.hs @@ -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 + diff --git a/hs-src/SecondTransfer/MainLoop/Framer.hs b/hs-src/SecondTransfer/MainLoop/Framer.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/MainLoop/Framer.hs @@ -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) + + diff --git a/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs b/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/MainLoop/OpenSSL_TLS.hs @@ -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 diff --git a/hs-src/SecondTransfer/MainLoop/PushPullType.hs b/hs-src/SecondTransfer/MainLoop/PushPullType.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/MainLoop/PushPullType.hs @@ -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 diff --git a/hs-src/SecondTransfer/MainLoop/Tokens.hs b/hs-src/SecondTransfer/MainLoop/Tokens.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/MainLoop/Tokens.hs @@ -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 diff --git a/hs-src/SecondTransfer/Utils.hs b/hs-src/SecondTransfer/Utils.hs new file mode 100644 --- /dev/null +++ b/hs-src/SecondTransfer/Utils.hs @@ -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 diff --git a/second-transfer.cabal b/second-transfer.cabal new file mode 100644 --- /dev/null +++ b/second-transfer.cabal @@ -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 diff --git a/tests/tests-hs-src/compiling_ok.hs b/tests/tests-hs-src/compiling_ok.hs new file mode 100644 --- /dev/null +++ b/tests/tests-hs-src/compiling_ok.hs @@ -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