diff --git a/src/AsyncSocket.h b/src/AsyncSocket.h index 4bdce35..979950c 100644 --- a/src/AsyncSocket.h +++ b/src/AsyncSocket.h @@ -9,6 +9,11 @@ namespace uWS { +// we need a variant of this where any failed write triggers a timeout! +// AsyncSocket needs to derive from a uSockets base we can select like so: + +// HttpResponse -> AsyncSocket -> TimeoutSocket or Socket (make StaticDispath Socket) + template struct AsyncSocket : StaticDispatch { template friend struct HttpContext; @@ -40,6 +45,11 @@ protected: return static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) this); } + bool isFullyOpen() { + // todo:: not shutdown or closed + return true; + } + /* Cork this socket. Only one socket may ever be corked per-loop at any given time */ void cork() { //std::cout << "Cork called" << std::endl; diff --git a/src/HttpContext.h b/src/HttpContext.h index c8c6aba..9a3ae3c 100644 --- a/src/HttpContext.h +++ b/src/HttpContext.h @@ -26,7 +26,7 @@ private: using StaticDispatch::static_dispatch; HttpContext() = delete; - /* Maximum delay allowed until an HTTP connection is terminated due to outstanding request (slow loris protection) */ + /* Maximum delay allowed until an HTTP connection is terminated due to outstanding request or rejected data (slow loris protection) */ static const int HTTP_IDLE_TIMEOUT_S = 10; SOCKET_CONTEXT_TYPE *getSocketContext() { @@ -95,22 +95,22 @@ private: return s; } - // basically, when getting a new request we need to disable timeouts and enter paused mode! + HttpResponseData *httpResponseData = (HttpResponseData *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s); /* Cork this socket */ ((AsyncSocket *) s)->cork(); - // pass this pointer to pointer along with the routing and change it if upgraded - SOCKET_TYPE *returnedSocket = s; + void *returnedSocket = httpResponseData->consumePostPadded(data, length, s, [httpContextData](void *s, uWS::HttpRequest *httpRequest) -> void * { - HttpResponseData *httpResponseData = (HttpResponseData *) static_dispatch(us_ssl_socket_ext, us_socket_ext)(s); - httpResponseData->consumePostPadded(data, length, s, [httpContextData](void *s, uWS::HttpRequest *httpRequest) { + // we need HttpAsyncSocket to derive from AsyncSocket where any failed write will trigger the timeout? - // whenever we get (the first) request we should disable timeout? + // http timeout logic in a nutshell: + // whenever a httpsocket writes and it fails, start a timer until onwritable, reset timer in next onwritable or next successful write + // if .end succeeds, then start a new timeout for the next request - // warning: if we are in shutdown state, resetting the timer is a security issue! - // todo: do not reset timer, disable it to allow hang requests! - static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) s, HTTP_IDLE_TIMEOUT_S); + /* For every request we reset the timeout and hang until user makes action */ + /* Warning: if we are in shutdown state, resetting the timer is a security issue! */ + static_dispatch(us_ssl_socket_timeout, us_socket_timeout)((SOCKET_TYPE *) s, 0); /* Reset httpResponse */ HttpResponseData *httpResponseData = (HttpResponseData *) static_dispatch(us_ssl_socket_ext, us_socket_ext)((SOCKET_TYPE *) s); @@ -118,32 +118,64 @@ private: httpResponseData->state = 0; /* Route the method and URL */ + + // I guess upgrade will have to write to a global variable we check afterwards httpContextData->router.route(httpRequest->getMethod(), httpRequest->getUrl(), { (HttpResponse *) s, httpRequest }); - // here we can be closed and in shutdown? - }, [httpResponseData](void *user, std::string_view data) { + //if (isFullyOpen) return s otherwise return nullptr; + + + /* Was the socket closed? */ + if (us_internal_socket_is_closed((struct us_socket *) s)) { + return nullptr; + } + + /* We absolutely have to terminate parsing if shutdown */ + if (static_dispatch(us_ssl_socket_is_shut_down, us_socket_is_shut_down)((SOCKET_TYPE *) s)) { + return nullptr; + } + + /* Was the socket upgraded? */ + // return new pointer to websocket + + /* If we return anything other than user, then that means STOP PARSING and this new value should be returned all the way? */ + + /* Continue parsing */ + return s; + + }, [httpResponseData](void *user, std::string_view data) -> void * { if (httpResponseData->inStream) { httpResponseData->inStream(data); } - }, [](void *user) { - // close any socket on HTTP errors - //static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) user); + /* Was the socket closed? */ + if (us_internal_socket_is_closed((struct us_socket *) user)) { + return nullptr; + } + + /* We absolutely have to terminate parsing if shutdown */ + if (static_dispatch(us_ssl_socket_is_shut_down, us_socket_is_shut_down)((SOCKET_TYPE *) user)) { + return nullptr; + } + + return user; + }, [](void *user) { + /* Close any socket on HTTP errors */ + static_dispatch(us_ssl_socket_close, us_socket_close)((SOCKET_TYPE *) user); + return nullptr; }); - if (us_internal_socket_is_closed((struct us_socket *) s)) { - // do you really return s? I guess so? - return s; + /* Only uncork still valid sockets */ + if (returnedSocket == s) { + ((AsyncSocket *) s)->uncork(); + } else { + // was this socket upgraded? + std::cout << "Socket was closed or shut down in handler or maybe upgraded" << std::endl; } - // uncork only if not closed - ((AsyncSocket *) s)->uncork(); - - // how do we return a new socket here, from the http route? - // maybe hold upgradedSocket in the loopData and return that? return s; }); @@ -152,8 +184,8 @@ private: std::cout << "HttpContext::onWritable event fired!" << std::endl; - /* Writing data should reset the timeout */ - static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, HTTP_IDLE_TIMEOUT_S); + /* We are now writable, so hang timeout again */ + static_dispatch(us_ssl_socket_timeout, us_socket_timeout)(s, 0); AsyncSocket *asyncSocket = (AsyncSocket *) s; HttpResponseData *httpResponseData = (HttpResponseData *) asyncSocket->getExt(); @@ -176,7 +208,7 @@ private: /* Handle FIN, HTTP does not support half-closed sockets, so simply close */ static_dispatch(us_ssl_socket_context_on_end, us_socket_context_on_end)(getSocketContext(), [](auto *s) { - std::cout << "FIN sent" << std::endl; + //std::cout << "FIN sent" << std::endl; /* We do not care for half closed sockets */ AsyncSocket *asyncSocket = (AsyncSocket *) s; @@ -226,15 +258,12 @@ public: static_dispatch(us_ssl_socket_context_free, us_socket_context_free)(getSocketContext()); } - /* Register an HTTP GET route handler acording to URL pattern */ + /* Register an HTTP route handler acording to URL pattern */ void onHttp(std::string method, std::string pattern, std::function *, uWS::HttpRequest *)> handler) { HttpContextData *httpContextData = getSocketContextData(); httpContextData->router.add(method, pattern, [handler](typename HttpContextData::RouterData user, std::pair params) { - - // todo: attach params to the req here! user.httpRequest->setParameters(params); - handler(user.httpResponse, user.httpRequest); }); } diff --git a/src/HttpParser.h b/src/HttpParser.h index 27651a6..e96de16 100644 --- a/src/HttpParser.h +++ b/src/HttpParser.h @@ -110,7 +110,7 @@ private: // the only caller of getHeaders template - int fenceAndConsumePostPadded(char *data, int length, void *user, HttpRequest *req, std::function &requestHandler, std::function &dataHandler) { + std::pair fenceAndConsumePostPadded(char *data, int length, void *user, HttpRequest *req, std::function &requestHandler, std::function &dataHandler) { int consumedTotal = 0; data[length] = '\r'; @@ -125,8 +125,19 @@ private: const char *querySeparatorPtr = (const char *) memchr(req->headers->value.data(), '?', req->headers->value.length()); req->querySeparator = (querySeparatorPtr ? querySeparatorPtr : req->headers->value.data() + req->headers->value.length()) - req->headers->value.data(); - requestHandler(user, req); + // this one should return socket and exit on closed + // what happens with data left for websockets? + void *returnedUser = requestHandler(user, req); + if (returnedUser != user) { + // upgraded socket, or otherwise broken + // return pair of consumed and user + return {consumedTotal, returnedUser}; + } + + // do not check this for GET! + + // todo: also support reading chunked streams std::string_view contentLengthString = req->getHeader("content-length"); if (contentLengthString.length()) { remainingStreamingBytes = toUnsignedInteger(contentLengthString); @@ -146,29 +157,36 @@ private: break; } } - return consumedTotal; + return {consumedTotal, user}; } public: // todo: what can we do with the socket inside the handlers? we need to check on return from any handler if we closed or terminated or upgraded the socket - void consumePostPadded(char *data, int length, void *user, std::function &&requestHandler, std::function &&dataHandler, std::function &&errorHandler) { + void *consumePostPadded(char *data, int length, void *user, std::function &&requestHandler, std::function &&dataHandler, std::function &&errorHandler) { HttpRequest req; if (remainingStreamingBytes) { + + // this is exactly the same as below! if (remainingStreamingBytes >= length) { - dataHandler(user, std::string_view(data, length)); + void *returnedUser = dataHandler(user, std::string_view(data, length)); remainingStreamingBytes -= length; - return; + return returnedUser; } else { - dataHandler(user, std::string_view(data, remainingStreamingBytes)); + void *returnedUser = dataHandler(user, std::string_view(data, remainingStreamingBytes)); data += remainingStreamingBytes; length -= remainingStreamingBytes; remainingStreamingBytes = 0; + + if (returnedUser != user) { + return returnedUser; + } } + } else if (fallback.length()) { int had = fallback.length(); @@ -177,50 +195,67 @@ public: fallback.reserve(maxCopyDistance + 32); // padding should be same as libus fallback.append(data, maxCopyDistance); - int consumed = fenceAndConsumePostPadded(fallback.data(), fallback.length(), user, &req, requestHandler, dataHandler); - if (consumed) { + // break here on break + std::pair consumed = fenceAndConsumePostPadded(fallback.data(), fallback.length(), user, &req, requestHandler, dataHandler); + if (consumed.second != user) { + return consumed.second; + } + + if (consumed.first) { fallback.clear(); - data += consumed - had; - length -= consumed - had; + data += consumed.first - had; + length -= consumed.first - had; - // this is exactly the same as above! if (remainingStreamingBytes) { + // this is exactly the same as above! if (remainingStreamingBytes >= length) { - dataHandler(user, std::string_view(data, length)); + void *returnedUser = dataHandler(user, std::string_view(data, length)); remainingStreamingBytes -= length; - return; + return returnedUser; } else { - dataHandler(user, std::string_view(data, remainingStreamingBytes)); + void *returnedUser = dataHandler(user, std::string_view(data, remainingStreamingBytes)); data += remainingStreamingBytes; length -= remainingStreamingBytes; remainingStreamingBytes = 0; + + if (returnedUser != user) { + return returnedUser; + } } } } else { if (fallback.length() == MAX_FALLBACK_SIZE) { - errorHandler(user); + // you don't really need error handler, just return something strange! + // we could have it return a constant pointer to denote error! + return errorHandler(user); } - return; + return user; } } - int consumed = fenceAndConsumePostPadded(data, length, user, &req, requestHandler, dataHandler); + std::pair consumed = fenceAndConsumePostPadded(data, length, user, &req, requestHandler, dataHandler); + if (consumed.second != user) { + return consumed.second; + } - data += consumed; - length -= consumed; + data += consumed.first; + length -= consumed.first; if (length) { if (length < MAX_FALLBACK_SIZE) { fallback.append(data, length); } else { - errorHandler(user); + return errorHandler(user); } } + + // added for now + return user; } }; diff --git a/src/HttpResponse.h b/src/HttpResponse.h index 1543775..3098c7a 100644 --- a/src/HttpResponse.h +++ b/src/HttpResponse.h @@ -103,6 +103,8 @@ public: Super::write(data.data(), data.length()); } + + // todo: calling end should start a timeout of the socket! } /* Write parts of the response in chunking fashion */