diff --git a/AK/AsyncStreamHelpers.h b/AK/AsyncStreamHelpers.h index bce3f2a2afa..3141cf56246 100644 --- a/AK/AsyncStreamHelpers.h +++ b/AK/AsyncStreamHelpers.h @@ -13,14 +13,17 @@ namespace AK { struct AsyncStreamHelpers { - static Coroutine> consume_until(AsyncInputStream& stream, StringView delimiter) + static Coroutine> consume_until(AsyncInputStream& stream, StringView delimiter, Optional max_size = {}) { size_t start_position = 0; while (true) { auto buffer = CO_TRY(co_await stream.peek()); auto position = StringUtils::find(StringView { buffer }, delimiter, start_position); - if (position.has_value()) - co_return must_sync(stream.read(*position + delimiter.length())); + auto end_position = position.map([&](auto value) { return value + delimiter.length(); }); + if (max_size.has_value() && !end_position.has_value() && buffer.size() >= max_size.value()) + end_position = buffer.size(); + if (end_position.has_value()) + co_return must_sync(stream.read(*end_position)); start_position = max(buffer.size(), delimiter.length() - 1) - delimiter.length() + 1; } } diff --git a/AK/Forward.h b/AK/Forward.h index 752e23ceb38..ad8b42c8349 100644 --- a/AK/Forward.h +++ b/AK/Forward.h @@ -38,6 +38,7 @@ class DeprecatedStringCodePointIterator; class Duration; class Error; class FlyString; +class GenericAwaiter; class GenericLexer; class IPv4Address; class JsonArray; diff --git a/AK/GenericAwaiter.h b/AK/GenericAwaiter.h new file mode 100644 index 00000000000..c4b9bf73c41 --- /dev/null +++ b/AK/GenericAwaiter.h @@ -0,0 +1,48 @@ +/* + * Copyright (c) 2024, Ali Mohammad Pur + * + * SPDX-License-Identifier: BSD-2-Clause + */ + +#pragma once + +#include +#include +#include + +namespace AK { +class GenericAwaiter { + AK_MAKE_NONMOVABLE(GenericAwaiter); + AK_MAKE_NONCOPYABLE(GenericAwaiter); + +public: + GenericAwaiter(Function)>&& fn) + { + fn([this] { ready(); }); + } + + bool await_ready() const { return false; } + void await_suspend(std::coroutine_handle<> handle) + { + m_handle = handle; + } + ErrorOr await_resume() + { + m_handle = {}; + return {}; + } + +private: + void ready() + { + if (m_handle) + m_handle.resume(); + } + + std::coroutine_handle<> m_handle; +}; +} + +#ifdef USING_AK_GLOBALLY +using AK::GenericAwaiter; +#endif diff --git a/Userland/Libraries/LibCore/File.h b/Userland/Libraries/LibCore/File.h index 23fc7626cb2..d9cc5f13c05 100644 --- a/Userland/Libraries/LibCore/File.h +++ b/Userland/Libraries/LibCore/File.h @@ -9,10 +9,13 @@ #include #include +#include +#include #include #include #include #include +#include #include namespace Core { @@ -77,6 +80,12 @@ public: // See also Socket::set_blocking. ErrorOr set_blocking(bool enabled); + Coroutine> wait_for_state(Core::Notifier::Type state) + { + auto notifier = CO_TRY(Core::Notifier::try_create(m_fd, state)); + co_return co_await GenericAwaiter([&](auto ready) { notifier->on_activation = move(ready); }); + } + template VIP> int leak_fd(Badge) { diff --git a/Userland/Libraries/LibCore/NetworkJob.cpp b/Userland/Libraries/LibCore/NetworkJob.cpp index 9ff275d8e69..9ec5161759b 100644 --- a/Userland/Libraries/LibCore/NetworkJob.cpp +++ b/Userland/Libraries/LibCore/NetworkJob.cpp @@ -11,7 +11,7 @@ namespace Core { -NetworkJob::NetworkJob(Stream& output_stream) +NetworkJob::NetworkJob(Core::File& output_stream) : m_output_stream(output_stream) { } diff --git a/Userland/Libraries/LibCore/NetworkJob.h b/Userland/Libraries/LibCore/NetworkJob.h index e8df3e38c9b..8102ade0575 100644 --- a/Userland/Libraries/LibCore/NetworkJob.h +++ b/Userland/Libraries/LibCore/NetworkJob.h @@ -7,6 +7,7 @@ #pragma once +#include #include #include #include @@ -53,16 +54,20 @@ public: } protected: - NetworkJob(Stream&); + NetworkJob(Core::File&); void did_finish(NonnullRefPtr&&); void did_fail(Error); void did_progress(Optional total_size, u64 downloaded); - ErrorOr do_write(ReadonlyBytes bytes) { return m_output_stream.write_some(bytes); } + Coroutine> do_write(ReadonlyBytes bytes) + { + CO_TRY(co_await m_output_stream.wait_for_state(Core::Notifier::Type::Write)); + co_return m_output_stream.write_some(bytes); + } private: RefPtr m_response; - Stream& m_output_stream; + Core::File& m_output_stream; Error m_error { Error::None }; }; diff --git a/Userland/Libraries/LibGemini/Job.cpp b/Userland/Libraries/LibGemini/Job.cpp index 578984687c2..603efdf20fe 100644 --- a/Userland/Libraries/LibGemini/Job.cpp +++ b/Userland/Libraries/LibGemini/Job.cpp @@ -8,13 +8,14 @@ #include #include #include +#include #include #include #include namespace Gemini { -Job::Job(GeminiRequest const& request, Stream& output_stream) +Job::Job(GeminiRequest const& request, Core::File& output_stream) : Core::NetworkJob(output_stream) , m_request(request) { @@ -83,7 +84,7 @@ void Job::flush_received_buffers() { for (size_t i = 0; i < m_received_buffers.size(); ++i) { auto& payload = m_received_buffers[i]; - auto result = do_write(payload); + auto result = Core::run_async_in_new_event_loop([&] { return do_write(payload); }); if (result.is_error()) { if (!result.error().is_errno()) { dbgln("Job: Failed to flush received buffers: {}", result.error()); diff --git a/Userland/Libraries/LibGemini/Job.h b/Userland/Libraries/LibGemini/Job.h index 89c23142cad..f966d128fdc 100644 --- a/Userland/Libraries/LibGemini/Job.h +++ b/Userland/Libraries/LibGemini/Job.h @@ -18,7 +18,7 @@ class Job : public Core::NetworkJob { C_OBJECT(Job); public: - explicit Job(GeminiRequest const&, Stream&); + explicit Job(GeminiRequest const&, Core::File&); virtual ~Job() override = default; virtual void start(Core::BufferedSocketBase&) override; diff --git a/Userland/Libraries/LibHTTP/HttpsJob.h b/Userland/Libraries/LibHTTP/HttpsJob.h index 0be336e0988..21b2e473851 100644 --- a/Userland/Libraries/LibHTTP/HttpsJob.h +++ b/Userland/Libraries/LibHTTP/HttpsJob.h @@ -30,7 +30,7 @@ public: Function()> on_certificate_requested; private: - explicit HttpsJob(HttpRequest&& request, Stream& output_stream) + explicit HttpsJob(HttpRequest&& request, Core::File& output_stream) : Job(move(request), output_stream) { } diff --git a/Userland/Libraries/LibHTTP/Job.cpp b/Userland/Libraries/LibHTTP/Job.cpp index 2e1074189d2..aa951463c85 100644 --- a/Userland/Libraries/LibHTTP/Job.cpp +++ b/Userland/Libraries/LibHTTP/Job.cpp @@ -5,14 +5,20 @@ * SPDX-License-Identifier: BSD-2-Clause */ +#include +#include +#include #include #include +#include +#include #include #include #include #include #include #include +#include #include #include #include @@ -20,6 +26,62 @@ namespace HTTP { +template +struct SyncStreamAsyncWrapper final : public AsyncInputStream { + explicit SyncStreamAsyncWrapper(MaybeOwned&& stream) + : m_stream(move(stream)) + , m_awaiter(GenericAwaiter { [&](auto cb) { m_stream->on_ready_to_read = move(cb); } }) + { + } + + virtual void reset() override + { + } + + virtual Coroutine> close() override + { + if (m_stream->is_open()) + m_stream->close(); + co_return {}; + } + + virtual bool is_open() const override + { + return !m_stream->is_eof() || !m_buffer.is_empty(); + } + + virtual Coroutine> enqueue_some(Badge) override + { + CO_TRY(co_await m_awaiter); + auto buffer = AK::Detail::ByteBuffer<16 * KiB> {}; + auto bytes = CO_TRY(m_stream->read_some(buffer.must_get_bytes_for_writing(16 * KiB))); + if (bytes.is_empty()) + co_return !m_stream->is_eof(); + m_buffer.append(bytes); + co_return true; + } + + virtual ReadonlyBytes buffered_data_unchecked(Badge) const override + { + return m_buffer.data(); + } + + virtual void dequeue(Badge, size_t bytes) override + { + m_buffer.dequeue(bytes); + } + + Coroutine> read_line(size_t max_size) + { + co_return StringView { CO_TRY(co_await AsyncStreamHelpers::consume_until(*this, "\r\n"sv, max_size)) }.trim("\r\n"sv, TrimMode::Right); + } + +private: + MaybeOwned m_stream; + GenericAwaiter m_awaiter; + AsyncStreamBuffer m_buffer; +}; + static ErrorOr handle_content_encoding(ByteBuffer const& buf, ByteString const& content_encoding) { dbgln_if(JOB_DEBUG, "Job::handle_content_encoding: buf has content_encoding={}", content_encoding); @@ -89,7 +151,7 @@ static ErrorOr handle_content_encoding(ByteBuffer const& buf, ByteSt return buf; } -Job::Job(HttpRequest&& request, Stream& output_stream) +Job::Job(HttpRequest&& request, Core::File& output_stream) : Core::NetworkJob(output_stream) , m_request(move(request)) { @@ -119,81 +181,335 @@ void Job::shutdown(ShutdownMode mode) } } -void Job::flush_received_buffers() +Coroutine Job::flush_received_buffers() { - if (!m_can_stream_response || m_buffered_size == 0) - return; - dbgln_if(JOB_DEBUG, "Job: Flushing received buffers: have {} bytes in {} buffers for {}", m_buffered_size, m_received_buffers.size(), m_request.url()); - for (size_t i = 0; i < m_received_buffers.size(); ++i) { - auto& payload = m_received_buffers[i]->pending_flush; - auto result = do_write(payload); - if (result.is_error()) { - if (!result.error().is_errno()) { - dbgln_if(JOB_DEBUG, "Job: Failed to flush received buffers: {}", result.error()); - continue; - } - if (result.error().code() == EINTR) { - i--; - continue; + if (!m_can_stream_response || m_has_scheduled_flush) + co_return; + + TemporaryChange change(m_has_scheduled_flush, true); + + while (m_buffered_size > 0) { + dbgln_if(JOB_DEBUG, "Job: Flushing received buffers: have {} bytes in {} buffers for {}", m_buffered_size, m_received_buffers.size(), m_request.url()); + for (size_t i = 0; i < m_received_buffers.size(); ++i) { + auto& payload = m_received_buffers[i]->pending_flush; + auto result = co_await do_write(payload); + if (result.is_error()) + break; + auto written = result.release_value(); + m_buffered_size -= written; + if (written == payload.size()) { + // FIXME: Make this a take-first-friendly object? + (void)m_received_buffers.take_first(); + break; } + VERIFY(written < payload.size()); + payload = payload.slice(written, payload.size() - written); break; } - auto written = result.release_value(); - m_buffered_size -= written; - if (written == payload.size()) { - // FIXME: Make this a take-first-friendly object? - (void)m_received_buffers.take_first(); - --i; - continue; - } - VERIFY(written < payload.size()); - payload = payload.slice(written, payload.size() - written); - break; + dbgln_if(JOB_DEBUG, "Job: Flushing received buffers done: have {} bytes in {} buffers for {}", m_buffered_size, m_received_buffers.size(), m_request.url()); + } + + if (m_request_done) { + m_has_scheduled_finish = true; + auto response = HttpResponse::create(m_code, move(m_headers), m_received_size); + // If the server responded with "Connection: close", close the connection + // as the server may or may not want to close the socket. Also, if this is + // a legacy HTTP server (1.0 or older), assume close is the default value. + if (auto result = response->headers().get("Connection"sv); result.has_value() ? result->equals_ignoring_ascii_case("close"sv) : m_legacy_connection) + shutdown(ShutdownMode::CloseSocket); + did_finish(response); } - dbgln_if(JOB_DEBUG, "Job: Flushing received buffers done: have {} bytes in {} buffers for {}", m_buffered_size, m_received_buffers.size(), m_request.url()); } -void Job::register_on_ready_to_read(Function callback) +auto Job::parse_status(auto& stream) -> Coroutine> { - m_socket->on_ready_to_read = [this, callback = move(callback)] { - callback(); + auto line = CO_TRY(co_await stream.read_line(PAGE_SIZE)); - // As `m_socket` is a buffered object, we might not get notifications for data in the buffer - // so exhaust the buffer to ensure we don't end up waiting forever. - auto can_read_without_blocking = m_socket->can_read_without_blocking(); - if (can_read_without_blocking.is_error()) - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - if (can_read_without_blocking.value() && m_state != State::Finished && !has_error()) { - deferred_invoke([this] { - if (m_socket && m_socket->on_ready_to_read) - m_socket->on_ready_to_read(); - }); + dbgln_if(JOB_DEBUG, "Job {} read line of length {}", m_request.url(), line.length()); + auto parts = line.split_view(' '); + if (parts.size() < 2) { + dbgln("Job: Expected 2-part or 3-part HTTP status line, got '{}'", line); + deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); }); + co_return {}; + } + + if (!parts[0].matches("HTTP/?.?"sv, CaseSensitivity::CaseSensitive) || !is_ascii_digit(parts[0][5]) || !is_ascii_digit(parts[0][7])) { + dbgln("Job: Expected HTTP-Version to be of the form 'HTTP/X.Y', got '{}'", parts[0]); + deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); }); + co_return {}; + } + auto http_major_version = parse_ascii_digit(parts[0][5]); + auto http_minor_version = parse_ascii_digit(parts[0][7]); + m_legacy_connection = http_major_version < 1 || (http_major_version == 1 && http_minor_version == 0); + + auto code = parts[1].template to_number(); + if (!code.has_value()) { + dbgln("Job: Expected numeric HTTP status"); + deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); }); + co_return {}; + } + m_code = code.value(); + co_return {}; +} + +auto Job::parse_headers(auto& stream, bool in_trailers) -> Coroutine> +{ + while (true) { + // There's no max limit defined on headers, but for our sanity, let's limit it to 32K. + auto line = StringView { CO_TRY(co_await stream.read_line(32 * KiB)) }; + + if (line.is_empty()) { + if (in_trailers) { + co_await finish_up(); + co_return {}; + } + if (on_headers_received) + on_headers_received(m_headers, m_code > 0 ? m_code : Optional {}); + + // We've reached the end of the headers, there's a possibility that the server + // responds with nothing (content-length = 0 with normal encoding); if that's the case, + // quit early as we won't be reading anything anyway. + if (auto result = m_headers.get("Content-Length"sv).value_or(""sv).to_number(); result.has_value()) { + if (result.value() == 0) { + auto transfer_encoding = m_headers.get("Transfer-Encoding"sv); + if (!transfer_encoding.has_value() || !transfer_encoding->view().trim_whitespace().equals_ignoring_ascii_case("chunked"sv)) { + co_await finish_up(); + co_return {}; + } + } + } + // There's also the possibility that the server responds with 204 (No Content), + // and manages to set a Content-Length anyway, in such cases ignore Content-Length and quit early; + // As the HTTP spec explicitly prohibits presence of Content-Length when the response code is 204. + if (m_code == 204) { + co_await finish_up(); + co_return {}; + } + + break; } - }; + auto parts = line.split_view(':'); + if (parts.is_empty()) { + if (in_trailers) { + // Some servers like to send two ending chunks + // use this fact as an excuse to ignore anything after the last chunk + // that is not a valid trailing header. + co_await finish_up(); + co_return {}; + } + dbgln("Job: Expected HTTP header with key/value"); + deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); }); + co_return {}; + } + auto name = parts[0]; + if (line.length() < name.length() + 2) { + if (in_trailers) { + // Some servers like to send two ending chunks + // use this fact as an excuse to ignore anything after the last chunk + // that is not a valid trailing header. + co_await finish_up(); + co_return {}; + } + dbgln("Job: Malformed HTTP header: '{}' ({})", line, line.length()); + deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); }); + co_return {}; + } + auto value = line.substring_view(name.length() + 2, line.length() - name.length() - 2); + m_headers.set(name, value); + if (name.equals_ignoring_ascii_case("Content-Encoding"sv)) { + // Assume that any content-encoding means that we can't decode it as a stream :( + dbgln_if(JOB_DEBUG, "Content-Encoding {} detected, cannot stream output :(", value); + m_can_stream_response = false; + } else if (name.equals_ignoring_ascii_case("Content-Length"sv)) { + auto length = value.to_number(); + if (length.has_value()) + m_content_length = length.value(); + } + dbgln_if(JOB_DEBUG, "Job: [{}] = '{}'", name, value); + } + + co_return {}; } -ErrorOr Job::read_line(size_t size) +auto Job::parse_body(auto& stream) -> Coroutine> { - auto buffer = TRY(ByteBuffer::create_uninitialized(size)); - auto bytes_read = TRY(m_socket->read_until(buffer, "\r\n"sv)); - return ByteString::copy(bytes_read); + auto should_read_trailers = false; + while (true) { + auto read_size = 64 * KiB; + if (m_current_chunk_remaining_size.has_value()) { + read_chunk_size:; + auto remaining = m_current_chunk_remaining_size.value(); + if (remaining == -1) { + // read size + auto size_data = CO_TRY(co_await stream.read_line(PAGE_SIZE)); + if (m_should_read_chunk_ending_line) { + // NOTE: Some servers seem to send an extra \r\n here despite there being no size. + // This makes us tolerate that. + size_data = size_data.trim("\r\n"sv, TrimMode::Right); + VERIFY(size_data.is_empty()); + m_should_read_chunk_ending_line = false; + continue; + } + auto size_lines = size_data.trim_whitespace().lines(); + dbgln_if(JOB_DEBUG, "Job: Received a chunk with size '{}'", size_data); + if (size_lines.size() == 0) { + if (!m_socket->is_eof()) + break; + dbgln("Job: Reached end of stream"); + co_await finish_up(); + break; + } + auto chunk = size_lines[0].split_view(';', SplitBehavior::KeepEmpty); + ByteString size_string = chunk[0]; + char* endptr; + auto size = strtoul(size_string.characters(), &endptr, 16); + if (*endptr) { + // invalid number + deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); + break; + } + if (size == 0) { + // This is the last chunk + // '0' *[; chunk-ext-name = chunk-ext-value] + // We're going to ignore _all_ chunk extensions + read_size = 0; + m_current_chunk_total_size = 0; + m_current_chunk_remaining_size = 0; + + dbgln_if(JOB_DEBUG, "Job: Received the last chunk with extensions '{}'", size_string.substring_view(1, size_string.length() - 1)); + } else { + m_current_chunk_total_size = size; + m_current_chunk_remaining_size = size; + read_size = size; + + dbgln_if(JOB_DEBUG, "Job: Chunk of size '{}' started", size); + } + } else { + read_size = remaining; + + dbgln_if(JOB_DEBUG, "Job: Resuming chunk with '{}' bytes left over", remaining); + } + } else { + auto transfer_encoding = m_headers.get("Transfer-Encoding"); + if (transfer_encoding.has_value()) { + // HTTP/1.1 3.3.3.3: + // If a message is received with both a Transfer-Encoding and a Content-Length header field, the Transfer-Encoding overrides the Content-Length. [...] + // https://httpwg.org/specs/rfc7230.html#message.body.length + m_content_length = {}; + + // Note: Some servers add extra spaces around 'chunked', see #6302. + auto encoding = transfer_encoding.value().trim_whitespace(); + + dbgln_if(JOB_DEBUG, "Job: This content has transfer encoding '{}'", encoding); + if (encoding.equals_ignoring_ascii_case("chunked"sv)) { + m_current_chunk_remaining_size = -1; + goto read_chunk_size; + } else { + dbgln("Job: Unknown transfer encoding '{}', the result will likely be wrong!", encoding); + } + } else if (m_content_length.has_value()) { + read_size = m_content_length.value() - m_received_size; + if (read_size == 0) { + co_await finish_up(); + break; + } + } + } + + auto payload = CO_TRY(ByteBuffer::copy(CO_TRY(co_await stream.read(read_size)))); + + if (payload.is_empty() && !stream.is_open()) { + co_await finish_up(); + break; + } + + bool read_everything = false; + if (m_content_length.has_value()) { + auto length = m_content_length.value(); + if (m_received_size + payload.size() >= length) { + payload.resize(length - m_received_size); + read_everything = true; + } + } + + m_received_buffers.append(make(payload)); + m_buffered_size += payload.size(); + m_received_size += payload.size(); + Core::EventLoop::current().adopt_coroutine(flush_received_buffers()); + + deferred_invoke([this] { did_progress(m_content_length, m_received_size); }); + + if (read_everything) { + VERIFY(m_received_size <= m_content_length.value()); + co_await finish_up(); + break; + } + + // Check after reading all the buffered data if we have reached the end of stream + // for cases where the server didn't send a content length, chunked encoding but is + // directly closing the connection. + if (!m_content_length.has_value() && !m_current_chunk_remaining_size.has_value() && m_socket->is_eof()) { + co_await finish_up(); + break; + } + + if (m_current_chunk_remaining_size.has_value()) { + auto size = m_current_chunk_remaining_size.value() - payload.size(); + + dbgln_if(JOB_DEBUG, "Job: We have {} bytes left over in this chunk", size); + if (size == 0) { + dbgln_if(JOB_DEBUG, "Job: Finished a chunk of {} bytes", m_current_chunk_total_size.value()); + + if (m_current_chunk_total_size.value() == 0) { + should_read_trailers = true; + break; + } + + // we've read everything, now let's get the next chunk + size = -1; + + auto line = CO_TRY(co_await stream.read_line(PAGE_SIZE)); + VERIFY(line.is_empty()); + } + m_current_chunk_remaining_size = size; + } + } + co_return should_read_trailers; } -ErrorOr Job::receive(size_t size) +auto Job::read_response() -> Coroutine> { - if (size == 0) - return ByteBuffer {}; + auto stream = SyncStreamAsyncWrapper(MaybeOwned { *m_socket }); - auto buffer = TRY(ByteBuffer::create_uninitialized(size)); - size_t nread; - do { - auto result = m_socket->read_some(buffer); - if (result.is_error() && result.error().is_errno() && result.error().code() == EINTR) - continue; - nread = TRY(result).size(); - break; - } while (true); - return buffer.slice(0, nread); + if (is_cancelled()) + co_return {}; + + if (m_socket->is_eof()) { + // Some servers really like terminating connections by simply closing them (even TLS ones) + // to signal end-of-data, if there's no: + // - connection + // - content-size + // - transfer-encoding: chunked + // header, simply treat EOF as a termination signal. + if (m_headers.contains("connection"sv) || m_content_length.has_value() || m_current_chunk_total_size.has_value()) { + dbgln_if(JOB_DEBUG, "Read failure: Actually EOF!"); + deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); }); + co_return {}; + } + + co_await finish_up(); + co_return {}; + } + + CO_TRY(co_await parse_status(stream)); + CO_TRY(co_await parse_headers(stream, false)); + auto should_read_trailers = CO_TRY(co_await parse_body(stream)); + if (should_read_trailers) + CO_TRY(co_await parse_headers(stream, true)); + + co_await finish_up(); + + co_return {}; } void Job::on_socket_connected() @@ -205,381 +521,27 @@ void Job::on_socket_connected() dbgln("{}", ByteString::copy(raw_request)); } - bool success = !m_socket->write_until_depleted(raw_request).is_error(); - if (!success) - deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); + if (m_socket->write_until_depleted(raw_request).is_error()) + return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - register_on_ready_to_read([&] { - dbgln_if(JOB_DEBUG, "Ready to read for {}, state = {}, cancelled = {}", m_request.url(), to_underlying(m_state), is_cancelled()); - if (is_cancelled()) - return; - - if (m_state == State::Finished) { - // We have everything we want, at this point, we can either get an EOF, or a bunch of extra newlines - // (unless "Connection: close" isn't specified) - // So just ignore everything after this. - return; + Core::EventLoop::current().adopt_coroutine([this] mutable -> Coroutine { + auto result = co_await read_response(); + if (result.is_error()) { + dbgln("Job: Failed to read response: {}", result.error()); + did_fail(Core::NetworkJob::Error::TransmissionFailed); } - - if (m_socket->is_eof()) { - // Some servers really like terminating connections by simply closing them (even TLS ones) - // to signal end-of-data, if there's no: - // - connection - // - content-size - // - transfer-encoding: chunked - // header, simply treat EOF as a termination signal. - if (m_headers.contains("connection"sv) || m_content_length.has_value() || m_current_chunk_total_size.has_value()) { - dbgln_if(JOB_DEBUG, "Read failure: Actually EOF!"); - deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); }); - return; - } - - finish_up(); - return; - } - - while (m_state == State::InStatus) { - auto can_read_line = m_socket->can_read_up_to_delimiter("\r\n"sv.bytes()); - if (can_read_line.is_error()) { - dbgln_if(JOB_DEBUG, "Job {} could not figure out whether we could read a line", m_request.url()); - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - } - - if (!can_read_line.value()) { - dbgln_if(JOB_DEBUG, "Job {} cannot read a full line", m_request.url()); - // TODO: Should we retry here instead of failing instantly? - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - } - - auto maybe_line = read_line(PAGE_SIZE); - if (maybe_line.is_error()) { - dbgln_if(JOB_DEBUG, "Job {} could not read line: {}", m_request.url(), maybe_line.error()); - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - } - - auto line = maybe_line.release_value(); - - dbgln_if(JOB_DEBUG, "Job {} read line of length {}", m_request.url(), line.length()); - auto parts = line.split_view(' '); - if (parts.size() < 2) { - dbgln("Job: Expected 2-part or 3-part HTTP status line, got '{}'", line); - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); }); - } - - if (!parts[0].matches("HTTP/?.?"sv, CaseSensitivity::CaseSensitive) || !is_ascii_digit(parts[0][5]) || !is_ascii_digit(parts[0][7])) { - dbgln("Job: Expected HTTP-Version to be of the form 'HTTP/X.Y', got '{}'", parts[0]); - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); }); - } - auto http_major_version = parse_ascii_digit(parts[0][5]); - auto http_minor_version = parse_ascii_digit(parts[0][7]); - m_legacy_connection = http_major_version < 1 || (http_major_version == 1 && http_minor_version == 0); - - auto code = parts[1].to_number(); - if (!code.has_value()) { - dbgln("Job: Expected numeric HTTP status"); - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); }); - } - m_code = code.value(); - m_state = State::InHeaders; - - auto can_read_without_blocking = m_socket->can_read_without_blocking(); - if (can_read_without_blocking.is_error()) - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - - if (!can_read_without_blocking.value()) - return; - } - while (m_state == State::InHeaders || m_state == State::Trailers) { - auto can_read_line = m_socket->can_read_up_to_delimiter("\r\n"sv.bytes()); - if (can_read_line.is_error()) { - dbgln_if(JOB_DEBUG, "Job {} could not figure out whether we could read a line", m_request.url()); - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - } - - if (!can_read_line.value()) { - dbgln_if(JOB_DEBUG, "Can't read lines anymore :("); - return; - } - - // There's no max limit defined on headers, but for our sanity, let's limit it to 32K. - auto maybe_line = read_line(32 * KiB); - if (maybe_line.is_error()) { - dbgln_if(JOB_DEBUG, "Job {} could not read a header line: {}", m_request.url(), maybe_line.error()); - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - } - auto line = maybe_line.release_value(); - - if (line.is_empty()) { - if (m_state == State::Trailers) { - return finish_up(); - } - if (on_headers_received) { - on_headers_received(m_headers, m_code > 0 ? m_code : Optional {}); - } - m_state = State::InBody; - - // We've reached the end of the headers, there's a possibility that the server - // responds with nothing (content-length = 0 with normal encoding); if that's the case, - // quit early as we won't be reading anything anyway. - if (auto result = m_headers.get("Content-Length"sv).value_or(""sv).to_number(); result.has_value()) { - if (result.value() == 0) { - auto transfer_encoding = m_headers.get("Transfer-Encoding"sv); - if (!transfer_encoding.has_value() || !transfer_encoding->view().trim_whitespace().equals_ignoring_ascii_case("chunked"sv)) - return finish_up(); - } - } - // There's also the possibility that the server responds with 204 (No Content), - // and manages to set a Content-Length anyway, in such cases ignore Content-Length and quit early; - // As the HTTP spec explicitly prohibits presence of Content-Length when the response code is 204. - if (m_code == 204) - return finish_up(); - - break; - } - auto parts = line.split_view(':'); - if (parts.is_empty()) { - if (m_state == State::Trailers) { - // Some servers like to send two ending chunks - // use this fact as an excuse to ignore anything after the last chunk - // that is not a valid trailing header. - return finish_up(); - } - dbgln("Job: Expected HTTP header with key/value"); - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); }); - } - auto name = parts[0]; - if (line.length() < name.length() + 2) { - if (m_state == State::Trailers) { - // Some servers like to send two ending chunks - // use this fact as an excuse to ignore anything after the last chunk - // that is not a valid trailing header. - return finish_up(); - } - dbgln("Job: Malformed HTTP header: '{}' ({})", line, line.length()); - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); }); - } - auto value = line.substring(name.length() + 2, line.length() - name.length() - 2); - m_headers.set(name, value); - if (name.equals_ignoring_ascii_case("Content-Encoding"sv)) { - // Assume that any content-encoding means that we can't decode it as a stream :( - dbgln_if(JOB_DEBUG, "Content-Encoding {} detected, cannot stream output :(", value); - m_can_stream_response = false; - } else if (name.equals_ignoring_ascii_case("Content-Length"sv)) { - auto length = value.to_number(); - if (length.has_value()) - m_content_length = length.value(); - } - dbgln_if(JOB_DEBUG, "Job: [{}] = '{}'", name, value); - - auto can_read_without_blocking = m_socket->can_read_without_blocking(); - if (can_read_without_blocking.is_error()) - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - if (!can_read_without_blocking.value()) { - dbgln_if(JOB_DEBUG, "Can't read headers anymore, byebye :("); - return; - } - } - VERIFY(m_state == State::InBody); - - while (true) { - auto can_read_without_blocking = m_socket->can_read_without_blocking(); - if (can_read_without_blocking.is_error()) - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - if (!can_read_without_blocking.value()) - break; - - auto read_size = 64 * KiB; - if (m_current_chunk_remaining_size.has_value()) { - read_chunk_size:; - auto remaining = m_current_chunk_remaining_size.value(); - if (remaining == -1) { - // read size - auto can_read_line = m_socket->can_read_up_to_delimiter("\r\n"sv.bytes()); - if (can_read_line.is_error()) - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - if (!can_read_line.value()) // We'll try later. - return; - auto maybe_size_data = read_line(PAGE_SIZE); - if (maybe_size_data.is_error()) { - dbgln_if(JOB_DEBUG, "Job: Could not receive chunk: {}", maybe_size_data.error()); - } - auto size_data = maybe_size_data.release_value(); - - if (m_should_read_chunk_ending_line) { - // NOTE: Some servers seem to send an extra \r\n here despite there being no size. - // This makes us tolerate that. - size_data = size_data.trim("\r\n"sv, TrimMode::Right); - VERIFY(size_data.is_empty()); - m_should_read_chunk_ending_line = false; - continue; - } - auto size_lines = size_data.view().trim_whitespace().lines(); - dbgln_if(JOB_DEBUG, "Job: Received a chunk with size '{}'", size_data); - if (size_lines.size() == 0) { - if (!m_socket->is_eof()) - break; - dbgln("Job: Reached end of stream"); - finish_up(); - break; - } else { - auto chunk = size_lines[0].split_view(';', SplitBehavior::KeepEmpty); - ByteString size_string = chunk[0]; - char* endptr; - auto size = strtoul(size_string.characters(), &endptr, 16); - if (*endptr) { - // invalid number - deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - break; - } - if (size == 0) { - // This is the last chunk - // '0' *[; chunk-ext-name = chunk-ext-value] - // We're going to ignore _all_ chunk extensions - read_size = 0; - m_current_chunk_total_size = 0; - m_current_chunk_remaining_size = 0; - - dbgln_if(JOB_DEBUG, "Job: Received the last chunk with extensions '{}'", size_string.substring_view(1, size_string.length() - 1)); - } else { - m_current_chunk_total_size = size; - m_current_chunk_remaining_size = size; - read_size = size; - - dbgln_if(JOB_DEBUG, "Job: Chunk of size '{}' started", size); - } - } - } else { - read_size = remaining; - - dbgln_if(JOB_DEBUG, "Job: Resuming chunk with '{}' bytes left over", remaining); - } - } else { - auto transfer_encoding = m_headers.get("Transfer-Encoding"); - if (transfer_encoding.has_value()) { - // HTTP/1.1 3.3.3.3: - // If a message is received with both a Transfer-Encoding and a Content-Length header field, the Transfer-Encoding overrides the Content-Length. [...] - // https://httpwg.org/specs/rfc7230.html#message.body.length - m_content_length = {}; - - // Note: Some servers add extra spaces around 'chunked', see #6302. - auto encoding = transfer_encoding.value().trim_whitespace(); - - dbgln_if(JOB_DEBUG, "Job: This content has transfer encoding '{}'", encoding); - if (encoding.equals_ignoring_ascii_case("chunked"sv)) { - m_current_chunk_remaining_size = -1; - goto read_chunk_size; - } else { - dbgln("Job: Unknown transfer encoding '{}', the result will likely be wrong!", encoding); - } - } - } - - can_read_without_blocking = m_socket->can_read_without_blocking(); - if (can_read_without_blocking.is_error()) - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - if (!can_read_without_blocking.value()) - break; - - dbgln_if(JOB_DEBUG, "Waiting for payload for {}", m_request.url()); - auto maybe_payload = receive(read_size); - if (maybe_payload.is_error()) { - dbgln_if(JOB_DEBUG, "Could not read the payload: {}", maybe_payload.error()); - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - } - - auto payload = maybe_payload.release_value(); - - if (payload.is_empty() && m_socket->is_eof()) { - finish_up(); - break; - } - - bool read_everything = false; - if (m_content_length.has_value()) { - auto length = m_content_length.value(); - if (m_received_size + payload.size() >= length) { - payload.resize(length - m_received_size); - read_everything = true; - } - } - - m_received_buffers.append(make(payload)); - m_buffered_size += payload.size(); - m_received_size += payload.size(); - flush_received_buffers(); - - deferred_invoke([this] { did_progress(m_content_length, m_received_size); }); - - if (read_everything) { - VERIFY(m_received_size <= m_content_length.value()); - finish_up(); - break; - } - - // Check after reading all the buffered data if we have reached the end of stream - // for cases where the server didn't send a content length, chunked encoding but is - // directly closing the connection. - if (!m_content_length.has_value() && !m_current_chunk_remaining_size.has_value() && m_socket->is_eof()) { - finish_up(); - break; - } - - if (m_current_chunk_remaining_size.has_value()) { - auto size = m_current_chunk_remaining_size.value() - payload.size(); - - dbgln_if(JOB_DEBUG, "Job: We have {} bytes left over in this chunk", size); - if (size == 0) { - dbgln_if(JOB_DEBUG, "Job: Finished a chunk of {} bytes", m_current_chunk_total_size.value()); - - if (m_current_chunk_total_size.value() == 0) { - m_state = State::Trailers; - break; - } - - // we've read everything, now let's get the next chunk - size = -1; - - auto can_read_line = m_socket->can_read_up_to_delimiter("\r\n"sv.bytes()); - if (can_read_line.is_error()) - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - if (can_read_line.value()) { - auto maybe_line = read_line(PAGE_SIZE); - if (maybe_line.is_error()) { - return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); }); - } - - VERIFY(maybe_line.value().is_empty()); - } else { - m_should_read_chunk_ending_line = true; - } - } - m_current_chunk_remaining_size = size; - } - } - - if (!m_socket->is_open()) { - dbgln_if(JOB_DEBUG, "Connection appears to have closed, finishing up"); - finish_up(); - } - }); + }()); } -void Job::timer_event(Core::TimerEvent& event) +Coroutine Job::finish_up() { - event.accept(); - finish_up(); - if (m_buffered_size == 0) - stop_timer(); -} + if (m_has_scheduled_finish) + co_return; -void Job::finish_up() -{ - VERIFY(!m_has_scheduled_finish); - m_state = State::Finished; if (!m_can_stream_response) { auto maybe_flattened_buffer = ByteBuffer::create_uninitialized(m_buffered_size); if (maybe_flattened_buffer.is_error()) - return did_fail(Core::NetworkJob::Error::TransmissionFailed); + co_return did_fail(Core::NetworkJob::Error::TransmissionFailed); auto flattened_buffer = maybe_flattened_buffer.release_value(); u8* flat_ptr = flattened_buffer.data(); @@ -596,7 +558,7 @@ void Job::finish_up() if (auto result = handle_content_encoding(flattened_buffer, content_encoding.value()); !result.is_error()) flattened_buffer = result.release_value(); else - return did_fail(Core::NetworkJob::Error::TransmissionFailed); + co_return did_fail(Core::NetworkJob::Error::TransmissionFailed); } m_buffered_size = flattened_buffer.size(); @@ -604,28 +566,7 @@ void Job::finish_up() m_can_stream_response = true; } - flush_received_buffers(); - if (m_buffered_size != 0) { - // We have to wait for the client to consume all the downloaded data - // before we can actually call `did_finish`. in a normal flow, this should - // never be hit since the client is reading as we are writing, unless there - // are too many concurrent downloads going on. - dbgln_if(JOB_DEBUG, "Flush finished with {} bytes remaining, will try again later", m_buffered_size); - if (!has_timer()) - start_timer(50); - return; - } - - stop_timer(); - m_has_scheduled_finish = true; - auto response = HttpResponse::create(m_code, move(m_headers), m_received_size); - deferred_invoke([this, response = move(response)] { - // If the server responded with "Connection: close", close the connection - // as the server may or may not want to close the socket. Also, if this is - // a legacy HTTP server (1.0 or older), assume close is the default value. - if (auto result = response->headers().get("Connection"sv); result.has_value() ? result->equals_ignoring_ascii_case("close"sv) : m_legacy_connection) - shutdown(ShutdownMode::CloseSocket); - did_finish(response); - }); + m_request_done = true; + Core::EventLoop::current().adopt_coroutine(flush_received_buffers()); } } diff --git a/Userland/Libraries/LibHTTP/Job.h b/Userland/Libraries/LibHTTP/Job.h index 15a18d5dbd9..19dd482ea7c 100644 --- a/Userland/Libraries/LibHTTP/Job.h +++ b/Userland/Libraries/LibHTTP/Job.h @@ -19,7 +19,7 @@ class Job : public Core::NetworkJob { C_OBJECT(Job); public: - explicit Job(HttpRequest&&, Stream&); + explicit Job(HttpRequest&&, Core::File&); virtual ~Job() override = default; virtual void start(Core::BufferedSocketBase&) override; @@ -31,25 +31,18 @@ public: HttpResponse* response() { return static_cast(Core::NetworkJob::response()); } HttpResponse const* response() const { return static_cast(Core::NetworkJob::response()); } -protected: - void finish_up(); - void on_socket_connected(); - void flush_received_buffers(); - void register_on_ready_to_read(Function); - ErrorOr read_line(size_t); - ErrorOr receive(size_t); - void timer_event(Core::TimerEvent&) override; +private: + auto parse_status(auto& stream) -> Coroutine>; + auto parse_headers(auto& stream, bool in_trailers) -> Coroutine>; + auto parse_body(auto& stream) -> Coroutine>; + auto read_response() -> Coroutine>; - enum class State { - InStatus, - InHeaders, - InBody, - Trailers, - Finished, - }; +protected: + Coroutine finish_up(); + void on_socket_connected(); + Coroutine flush_received_buffers(); HttpRequest m_request; - State m_state { State::InStatus }; Core::BufferedSocketBase* m_socket { nullptr }; bool m_legacy_connection { false }; int m_code { -1 }; @@ -79,6 +72,8 @@ protected: bool m_can_stream_response { true }; bool m_should_read_chunk_ending_line { false }; bool m_has_scheduled_finish { false }; + bool m_has_scheduled_flush { false }; + bool m_request_done { false }; }; } diff --git a/Userland/Services/RequestServer/ConnectionCache.h b/Userland/Services/RequestServer/ConnectionCache.h index 1a0aaa56454..20abdba1dcd 100644 --- a/Userland/Services/RequestServer/ConnectionCache.h +++ b/Userland/Services/RequestServer/ConnectionCache.h @@ -20,6 +20,9 @@ #include #include +#undef REQUESTSERVER_DEBUG +#define REQUESTSERVER_DEBUG 0 + namespace RequestServer { enum class CacheLevel {