Job.cpp 28 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630
  1. /*
  2. * Copyright (c) 2018-2020, Andreas Kling <kling@serenityos.org>
  3. * Copyright (c) 2022, the SerenityOS developers.
  4. *
  5. * SPDX-License-Identifier: BSD-2-Clause
  6. */
  7. #include <AK/CharacterTypes.h>
  8. #include <AK/Debug.h>
  9. #include <AK/JsonObject.h>
  10. #include <AK/MemoryStream.h>
  11. #include <AK/Try.h>
  12. #include <LibCompress/Brotli.h>
  13. #include <LibCompress/Gzip.h>
  14. #include <LibCompress/Zlib.h>
  15. #include <LibCore/Event.h>
  16. #include <LibHTTP/HttpResponse.h>
  17. #include <LibHTTP/Job.h>
  18. #include <stdio.h>
  19. #include <unistd.h>
  20. namespace HTTP {
  21. static ErrorOr<ByteBuffer> handle_content_encoding(ByteBuffer const& buf, DeprecatedString const& content_encoding)
  22. {
  23. dbgln_if(JOB_DEBUG, "Job::handle_content_encoding: buf has content_encoding={}", content_encoding);
  24. // FIXME: Actually do the decompression of the data using streams, instead of all at once when everything has been
  25. // received. This will require that some of the decompression algorithms are implemented in a streaming way.
  26. // Gzip and Deflate are implemented using Stream, while Brotli uses the newer Core::Stream. The Gzip and
  27. // Deflate implementations will likely need to be changed to LibCore::Stream for this to work easily.
  28. if (content_encoding == "gzip") {
  29. if (!Compress::GzipDecompressor::is_likely_compressed(buf)) {
  30. dbgln("Job::handle_content_encoding: buf is not gzip compressed!");
  31. }
  32. dbgln_if(JOB_DEBUG, "Job::handle_content_encoding: buf is gzip compressed!");
  33. auto uncompressed = TRY(Compress::GzipDecompressor::decompress_all(buf));
  34. if constexpr (JOB_DEBUG) {
  35. dbgln("Job::handle_content_encoding: Gzip::decompress() successful.");
  36. dbgln(" Input size: {}", buf.size());
  37. dbgln(" Output size: {}", uncompressed.size());
  38. }
  39. return uncompressed;
  40. } else if (content_encoding == "deflate") {
  41. dbgln_if(JOB_DEBUG, "Job::handle_content_encoding: buf is deflate compressed!");
  42. // Even though the content encoding is "deflate", it's actually deflate with the zlib wrapper.
  43. // https://tools.ietf.org/html/rfc7230#section-4.2.2
  44. auto memory_stream = make<FixedMemoryStream>(buf);
  45. auto zlib_decompressor = Compress::ZlibDecompressor::create(move(memory_stream));
  46. Optional<ByteBuffer> uncompressed;
  47. if (zlib_decompressor.is_error()) {
  48. // From the RFC:
  49. // "Note: Some non-conformant implementations send the "deflate"
  50. // compressed data without the zlib wrapper."
  51. dbgln_if(JOB_DEBUG, "Job::handle_content_encoding: ZlibDecompressor::decompress_all() failed. Trying DeflateDecompressor::decompress_all()");
  52. uncompressed = TRY(Compress::DeflateDecompressor::decompress_all(buf));
  53. } else {
  54. uncompressed = TRY(zlib_decompressor.value()->read_until_eof());
  55. }
  56. if constexpr (JOB_DEBUG) {
  57. dbgln("Job::handle_content_encoding: Deflate decompression successful.");
  58. dbgln(" Input size: {}", buf.size());
  59. dbgln(" Output size: {}", uncompressed.value().size());
  60. }
  61. return uncompressed.release_value();
  62. } else if (content_encoding == "br") {
  63. dbgln_if(JOB_DEBUG, "Job::handle_content_encoding: buf is brotli compressed!");
  64. FixedMemoryStream bufstream { buf };
  65. auto brotli_stream = Compress::BrotliDecompressionStream { MaybeOwned<Stream>(bufstream) };
  66. auto uncompressed = TRY(brotli_stream.read_until_eof());
  67. if constexpr (JOB_DEBUG) {
  68. dbgln("Job::handle_content_encoding: Brotli::decompress() successful.");
  69. dbgln(" Input size: {}", buf.size());
  70. dbgln(" Output size: {}", uncompressed.size());
  71. }
  72. return uncompressed;
  73. }
  74. return buf;
  75. }
  76. Job::Job(HttpRequest&& request, Stream& output_stream)
  77. : Core::NetworkJob(output_stream)
  78. , m_request(move(request))
  79. {
  80. }
  81. void Job::start(Core::BufferedSocketBase& socket)
  82. {
  83. VERIFY(!m_socket);
  84. m_socket = &socket;
  85. dbgln_if(HTTPJOB_DEBUG, "Reusing previous connection for {}", url());
  86. deferred_invoke([this] {
  87. dbgln_if(HTTPJOB_DEBUG, "HttpJob: on_connected callback");
  88. on_socket_connected();
  89. });
  90. }
  91. void Job::shutdown(ShutdownMode mode)
  92. {
  93. if (!m_socket)
  94. return;
  95. if (mode == ShutdownMode::CloseSocket) {
  96. m_socket->close();
  97. m_socket->on_ready_to_read = nullptr;
  98. } else {
  99. m_socket->on_ready_to_read = nullptr;
  100. m_socket = nullptr;
  101. }
  102. }
  103. void Job::flush_received_buffers()
  104. {
  105. if (!m_can_stream_response || m_buffered_size == 0)
  106. return;
  107. dbgln_if(JOB_DEBUG, "Job: Flushing received buffers: have {} bytes in {} buffers for {}", m_buffered_size, m_received_buffers.size(), m_request.url());
  108. for (size_t i = 0; i < m_received_buffers.size(); ++i) {
  109. auto& payload = m_received_buffers[i]->pending_flush;
  110. auto result = do_write(payload);
  111. if (result.is_error()) {
  112. if (!result.error().is_errno()) {
  113. dbgln_if(JOB_DEBUG, "Job: Failed to flush received buffers: {}", result.error());
  114. continue;
  115. }
  116. if (result.error().code() == EINTR) {
  117. i--;
  118. continue;
  119. }
  120. break;
  121. }
  122. auto written = result.release_value();
  123. m_buffered_size -= written;
  124. if (written == payload.size()) {
  125. // FIXME: Make this a take-first-friendly object?
  126. (void)m_received_buffers.take_first();
  127. --i;
  128. continue;
  129. }
  130. VERIFY(written < payload.size());
  131. payload = payload.slice(written, payload.size() - written);
  132. break;
  133. }
  134. dbgln_if(JOB_DEBUG, "Job: Flushing received buffers done: have {} bytes in {} buffers for {}", m_buffered_size, m_received_buffers.size(), m_request.url());
  135. }
  136. void Job::register_on_ready_to_read(Function<void()> callback)
  137. {
  138. m_socket->on_ready_to_read = [this, callback = move(callback)] {
  139. callback();
  140. // As `m_socket` is a buffered object, we might not get notifications for data in the buffer
  141. // so exhaust the buffer to ensure we don't end up waiting forever.
  142. auto can_read_without_blocking = m_socket->can_read_without_blocking();
  143. if (can_read_without_blocking.is_error())
  144. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  145. if (can_read_without_blocking.value() && m_state != State::Finished && !has_error()) {
  146. deferred_invoke([this] {
  147. if (m_socket && m_socket->on_ready_to_read)
  148. m_socket->on_ready_to_read();
  149. });
  150. }
  151. };
  152. }
  153. ErrorOr<DeprecatedString> Job::read_line(size_t size)
  154. {
  155. auto buffer = TRY(ByteBuffer::create_uninitialized(size));
  156. auto bytes_read = TRY(m_socket->read_until(buffer, "\r\n"sv));
  157. return DeprecatedString::copy(bytes_read);
  158. }
  159. ErrorOr<ByteBuffer> Job::receive(size_t size)
  160. {
  161. if (size == 0)
  162. return ByteBuffer {};
  163. auto buffer = TRY(ByteBuffer::create_uninitialized(size));
  164. size_t nread;
  165. do {
  166. auto result = m_socket->read_some(buffer);
  167. if (result.is_error() && result.error().is_errno() && result.error().code() == EINTR)
  168. continue;
  169. nread = TRY(result).size();
  170. break;
  171. } while (true);
  172. return buffer.slice(0, nread);
  173. }
  174. void Job::on_socket_connected()
  175. {
  176. auto raw_request = m_request.to_raw_request().release_value_but_fixme_should_propagate_errors();
  177. if constexpr (JOB_DEBUG) {
  178. dbgln("Job: raw_request:");
  179. dbgln("{}", DeprecatedString::copy(raw_request));
  180. }
  181. bool success = !m_socket->write_until_depleted(raw_request).is_error();
  182. if (!success)
  183. deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  184. register_on_ready_to_read([&] {
  185. dbgln_if(JOB_DEBUG, "Ready to read for {}, state = {}, cancelled = {}", m_request.url(), to_underlying(m_state), is_cancelled());
  186. if (is_cancelled())
  187. return;
  188. if (m_state == State::Finished) {
  189. // We have everything we want, at this point, we can either get an EOF, or a bunch of extra newlines
  190. // (unless "Connection: close" isn't specified)
  191. // So just ignore everything after this.
  192. return;
  193. }
  194. if (m_socket->is_eof()) {
  195. dbgln_if(JOB_DEBUG, "Read failure: Actually EOF!");
  196. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
  197. }
  198. while (m_state == State::InStatus) {
  199. auto can_read_line = m_socket->can_read_line();
  200. if (can_read_line.is_error()) {
  201. dbgln_if(JOB_DEBUG, "Job {} could not figure out whether we could read a line", m_request.url());
  202. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  203. }
  204. if (!can_read_line.value()) {
  205. dbgln_if(JOB_DEBUG, "Job {} cannot read a full line", m_request.url());
  206. // TODO: Should we retry here instead of failing instantly?
  207. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  208. }
  209. auto maybe_line = read_line(PAGE_SIZE);
  210. if (maybe_line.is_error()) {
  211. dbgln_if(JOB_DEBUG, "Job {} could not read line: {}", m_request.url(), maybe_line.error());
  212. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  213. }
  214. auto line = maybe_line.release_value();
  215. dbgln_if(JOB_DEBUG, "Job {} read line of length {}", m_request.url(), line.length());
  216. auto parts = line.split_view(' ');
  217. if (parts.size() < 2) {
  218. dbgln("Job: Expected 2-part or 3-part HTTP status line, got '{}'", line);
  219. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
  220. }
  221. if (!parts[0].matches("HTTP/?.?"sv, CaseSensitivity::CaseSensitive) || !is_ascii_digit(parts[0][5]) || !is_ascii_digit(parts[0][7])) {
  222. dbgln("Job: Expected HTTP-Version to be of the form 'HTTP/X.Y', got '{}'", parts[0]);
  223. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
  224. }
  225. auto http_major_version = parse_ascii_digit(parts[0][5]);
  226. auto http_minor_version = parse_ascii_digit(parts[0][7]);
  227. m_legacy_connection = http_major_version < 1 || (http_major_version == 1 && http_minor_version == 0);
  228. auto code = parts[1].to_uint();
  229. if (!code.has_value()) {
  230. dbgln("Job: Expected numeric HTTP status");
  231. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
  232. }
  233. m_code = code.value();
  234. m_state = State::InHeaders;
  235. auto can_read_without_blocking = m_socket->can_read_without_blocking();
  236. if (can_read_without_blocking.is_error())
  237. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  238. if (!can_read_without_blocking.value())
  239. return;
  240. }
  241. while (m_state == State::InHeaders || m_state == State::Trailers) {
  242. auto can_read_line = m_socket->can_read_line();
  243. if (can_read_line.is_error()) {
  244. dbgln_if(JOB_DEBUG, "Job {} could not figure out whether we could read a line", m_request.url());
  245. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  246. }
  247. if (!can_read_line.value()) {
  248. dbgln_if(JOB_DEBUG, "Can't read lines anymore :(");
  249. return;
  250. }
  251. // There's no max limit defined on headers, but for our sanity, let's limit it to 32K.
  252. auto maybe_line = read_line(32 * KiB);
  253. if (maybe_line.is_error()) {
  254. dbgln_if(JOB_DEBUG, "Job {} could not read a header line: {}", m_request.url(), maybe_line.error());
  255. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  256. }
  257. auto line = maybe_line.release_value();
  258. if (line.is_empty()) {
  259. if (m_state == State::Trailers) {
  260. return finish_up();
  261. }
  262. if (on_headers_received) {
  263. if (!m_set_cookie_headers.is_empty())
  264. m_headers.set("Set-Cookie", JsonArray { m_set_cookie_headers }.to_deprecated_string());
  265. on_headers_received(m_headers, m_code > 0 ? m_code : Optional<u32> {});
  266. }
  267. m_state = State::InBody;
  268. // We've reached the end of the headers, there's a possibility that the server
  269. // responds with nothing (content-length = 0 with normal encoding); if that's the case,
  270. // quit early as we won't be reading anything anyway.
  271. if (auto result = m_headers.get("Content-Length"sv).value_or(""sv).to_uint(); result.has_value()) {
  272. if (result.value() == 0 && !m_headers.get("Transfer-Encoding"sv).value_or(""sv).view().trim_whitespace().equals_ignoring_ascii_case("chunked"sv))
  273. return finish_up();
  274. }
  275. // There's also the possibility that the server responds with 204 (No Content),
  276. // and manages to set a Content-Length anyway, in such cases ignore Content-Length and quit early;
  277. // As the HTTP spec explicitly prohibits presence of Content-Length when the response code is 204.
  278. if (m_code == 204)
  279. return finish_up();
  280. break;
  281. }
  282. auto parts = line.split_view(':');
  283. if (parts.is_empty()) {
  284. if (m_state == State::Trailers) {
  285. // Some servers like to send two ending chunks
  286. // use this fact as an excuse to ignore anything after the last chunk
  287. // that is not a valid trailing header.
  288. return finish_up();
  289. }
  290. dbgln("Job: Expected HTTP header with key/value");
  291. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
  292. }
  293. auto name = parts[0];
  294. if (line.length() < name.length() + 2) {
  295. if (m_state == State::Trailers) {
  296. // Some servers like to send two ending chunks
  297. // use this fact as an excuse to ignore anything after the last chunk
  298. // that is not a valid trailing header.
  299. return finish_up();
  300. }
  301. dbgln("Job: Malformed HTTP header: '{}' ({})", line, line.length());
  302. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::ProtocolFailed); });
  303. }
  304. auto value = line.substring(name.length() + 2, line.length() - name.length() - 2);
  305. if (name.equals_ignoring_ascii_case("Set-Cookie"sv)) {
  306. dbgln_if(JOB_DEBUG, "Job: Received Set-Cookie header: '{}'", value);
  307. m_set_cookie_headers.append(move(value));
  308. auto can_read_without_blocking = m_socket->can_read_without_blocking();
  309. if (can_read_without_blocking.is_error())
  310. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  311. if (!can_read_without_blocking.value())
  312. return;
  313. } else if (auto existing_value = m_headers.get(name); existing_value.has_value()) {
  314. StringBuilder builder;
  315. builder.append(existing_value.value());
  316. builder.append(',');
  317. builder.append(value);
  318. m_headers.set(name, builder.to_deprecated_string());
  319. } else {
  320. m_headers.set(name, value);
  321. }
  322. if (name.equals_ignoring_ascii_case("Content-Encoding"sv)) {
  323. // Assume that any content-encoding means that we can't decode it as a stream :(
  324. dbgln_if(JOB_DEBUG, "Content-Encoding {} detected, cannot stream output :(", value);
  325. m_can_stream_response = false;
  326. } else if (name.equals_ignoring_ascii_case("Content-Length"sv)) {
  327. auto length = value.to_uint<u64>();
  328. if (length.has_value())
  329. m_content_length = length.value();
  330. }
  331. dbgln_if(JOB_DEBUG, "Job: [{}] = '{}'", name, value);
  332. auto can_read_without_blocking = m_socket->can_read_without_blocking();
  333. if (can_read_without_blocking.is_error())
  334. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  335. if (!can_read_without_blocking.value()) {
  336. dbgln_if(JOB_DEBUG, "Can't read headers anymore, byebye :(");
  337. return;
  338. }
  339. }
  340. VERIFY(m_state == State::InBody);
  341. while (true) {
  342. auto can_read_without_blocking = m_socket->can_read_without_blocking();
  343. if (can_read_without_blocking.is_error())
  344. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  345. if (!can_read_without_blocking.value())
  346. break;
  347. auto read_size = 64 * KiB;
  348. if (m_current_chunk_remaining_size.has_value()) {
  349. read_chunk_size:;
  350. auto remaining = m_current_chunk_remaining_size.value();
  351. if (remaining == -1) {
  352. // read size
  353. auto maybe_size_data = read_line(PAGE_SIZE);
  354. if (maybe_size_data.is_error()) {
  355. dbgln_if(JOB_DEBUG, "Job: Could not receive chunk: {}", maybe_size_data.error());
  356. }
  357. auto size_data = maybe_size_data.release_value();
  358. if (m_should_read_chunk_ending_line) {
  359. // NOTE: Some servers seem to send an extra \r\n here despite there being no size.
  360. // This makes us tolerate that.
  361. size_data = size_data.trim("\r\n"sv, TrimMode::Right);
  362. VERIFY(size_data.is_empty());
  363. m_should_read_chunk_ending_line = false;
  364. continue;
  365. }
  366. auto size_lines = size_data.view().lines();
  367. dbgln_if(JOB_DEBUG, "Job: Received a chunk with size '{}'", size_data);
  368. if (size_lines.size() == 0) {
  369. if (!m_socket->is_eof())
  370. break;
  371. dbgln("Job: Reached end of stream");
  372. finish_up();
  373. break;
  374. } else {
  375. auto chunk = size_lines[0].split_view(';', SplitBehavior::KeepEmpty);
  376. DeprecatedString size_string = chunk[0];
  377. char* endptr;
  378. auto size = strtoul(size_string.characters(), &endptr, 16);
  379. if (*endptr) {
  380. // invalid number
  381. deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  382. break;
  383. }
  384. if (size == 0) {
  385. // This is the last chunk
  386. // '0' *[; chunk-ext-name = chunk-ext-value]
  387. // We're going to ignore _all_ chunk extensions
  388. read_size = 0;
  389. m_current_chunk_total_size = 0;
  390. m_current_chunk_remaining_size = 0;
  391. dbgln_if(JOB_DEBUG, "Job: Received the last chunk with extensions '{}'", size_string.substring_view(1, size_string.length() - 1));
  392. } else {
  393. m_current_chunk_total_size = size;
  394. m_current_chunk_remaining_size = size;
  395. read_size = size;
  396. dbgln_if(JOB_DEBUG, "Job: Chunk of size '{}' started", size);
  397. }
  398. }
  399. } else {
  400. read_size = remaining;
  401. dbgln_if(JOB_DEBUG, "Job: Resuming chunk with '{}' bytes left over", remaining);
  402. }
  403. } else {
  404. auto transfer_encoding = m_headers.get("Transfer-Encoding");
  405. if (transfer_encoding.has_value()) {
  406. // HTTP/1.1 3.3.3.3:
  407. // If a message is received with both a Transfer-Encoding and a Content-Length header field, the Transfer-Encoding overrides the Content-Length. [...]
  408. // https://httpwg.org/specs/rfc7230.html#message.body.length
  409. m_content_length = {};
  410. // Note: Some servers add extra spaces around 'chunked', see #6302.
  411. auto encoding = transfer_encoding.value().trim_whitespace();
  412. dbgln_if(JOB_DEBUG, "Job: This content has transfer encoding '{}'", encoding);
  413. if (encoding.equals_ignoring_ascii_case("chunked"sv)) {
  414. m_current_chunk_remaining_size = -1;
  415. goto read_chunk_size;
  416. } else {
  417. dbgln("Job: Unknown transfer encoding '{}', the result will likely be wrong!", encoding);
  418. }
  419. }
  420. }
  421. can_read_without_blocking = m_socket->can_read_without_blocking();
  422. if (can_read_without_blocking.is_error())
  423. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  424. if (!can_read_without_blocking.value())
  425. break;
  426. dbgln_if(JOB_DEBUG, "Waiting for payload for {}", m_request.url());
  427. auto maybe_payload = receive(read_size);
  428. if (maybe_payload.is_error()) {
  429. dbgln_if(JOB_DEBUG, "Could not read the payload: {}", maybe_payload.error());
  430. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  431. }
  432. auto payload = maybe_payload.release_value();
  433. if (payload.is_empty() && m_socket->is_eof()) {
  434. finish_up();
  435. break;
  436. }
  437. bool read_everything = false;
  438. if (m_content_length.has_value()) {
  439. auto length = m_content_length.value();
  440. if (m_received_size + payload.size() >= length) {
  441. payload.resize(length - m_received_size);
  442. read_everything = true;
  443. }
  444. }
  445. m_received_buffers.append(make<ReceivedBuffer>(payload));
  446. m_buffered_size += payload.size();
  447. m_received_size += payload.size();
  448. flush_received_buffers();
  449. deferred_invoke([this] { did_progress(m_content_length, m_received_size); });
  450. if (read_everything) {
  451. VERIFY(m_received_size <= m_content_length.value());
  452. finish_up();
  453. break;
  454. }
  455. // Check after reading all the buffered data if we have reached the end of stream
  456. // for cases where the server didn't send a content length, chunked encoding but is
  457. // directly closing the connection.
  458. if (!m_content_length.has_value() && !m_current_chunk_remaining_size.has_value() && m_socket->is_eof()) {
  459. finish_up();
  460. break;
  461. }
  462. if (m_current_chunk_remaining_size.has_value()) {
  463. auto size = m_current_chunk_remaining_size.value() - payload.size();
  464. dbgln_if(JOB_DEBUG, "Job: We have {} bytes left over in this chunk", size);
  465. if (size == 0) {
  466. dbgln_if(JOB_DEBUG, "Job: Finished a chunk of {} bytes", m_current_chunk_total_size.value());
  467. if (m_current_chunk_total_size.value() == 0) {
  468. m_state = State::Trailers;
  469. break;
  470. }
  471. // we've read everything, now let's get the next chunk
  472. size = -1;
  473. auto can_read_line = m_socket->can_read_line();
  474. if (can_read_line.is_error())
  475. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  476. if (can_read_line.value()) {
  477. auto maybe_line = read_line(PAGE_SIZE);
  478. if (maybe_line.is_error()) {
  479. return deferred_invoke([this] { did_fail(Core::NetworkJob::Error::TransmissionFailed); });
  480. }
  481. VERIFY(maybe_line.value().is_empty());
  482. } else {
  483. m_should_read_chunk_ending_line = true;
  484. }
  485. }
  486. m_current_chunk_remaining_size = size;
  487. }
  488. }
  489. if (!m_socket->is_open()) {
  490. dbgln_if(JOB_DEBUG, "Connection appears to have closed, finishing up");
  491. finish_up();
  492. }
  493. });
  494. }
  495. void Job::timer_event(Core::TimerEvent& event)
  496. {
  497. event.accept();
  498. finish_up();
  499. if (m_buffered_size == 0)
  500. stop_timer();
  501. }
  502. void Job::finish_up()
  503. {
  504. VERIFY(!m_has_scheduled_finish);
  505. m_state = State::Finished;
  506. if (!m_can_stream_response) {
  507. auto maybe_flattened_buffer = ByteBuffer::create_uninitialized(m_buffered_size);
  508. if (maybe_flattened_buffer.is_error())
  509. return did_fail(Core::NetworkJob::Error::TransmissionFailed);
  510. auto flattened_buffer = maybe_flattened_buffer.release_value();
  511. u8* flat_ptr = flattened_buffer.data();
  512. for (auto& received_buffer : m_received_buffers) {
  513. memcpy(flat_ptr, received_buffer->pending_flush.data(), received_buffer->pending_flush.size());
  514. flat_ptr += received_buffer->pending_flush.size();
  515. }
  516. m_received_buffers.clear();
  517. // For the time being, we cannot stream stuff with content-encoding set to _anything_.
  518. // FIXME: LibCompress exposes a streaming interface, so this can be resolved
  519. auto content_encoding = m_headers.get("Content-Encoding");
  520. if (content_encoding.has_value()) {
  521. if (auto result = handle_content_encoding(flattened_buffer, content_encoding.value()); !result.is_error())
  522. flattened_buffer = result.release_value();
  523. else
  524. return did_fail(Core::NetworkJob::Error::TransmissionFailed);
  525. }
  526. m_buffered_size = flattened_buffer.size();
  527. m_received_buffers.append(make<ReceivedBuffer>(move(flattened_buffer)));
  528. m_can_stream_response = true;
  529. }
  530. flush_received_buffers();
  531. if (m_buffered_size != 0) {
  532. // We have to wait for the client to consume all the downloaded data
  533. // before we can actually call `did_finish`. in a normal flow, this should
  534. // never be hit since the client is reading as we are writing, unless there
  535. // are too many concurrent downloads going on.
  536. dbgln_if(JOB_DEBUG, "Flush finished with {} bytes remaining, will try again later", m_buffered_size);
  537. if (!has_timer())
  538. start_timer(50);
  539. return;
  540. }
  541. m_has_scheduled_finish = true;
  542. auto response = HttpResponse::create(m_code, move(m_headers), m_received_size);
  543. deferred_invoke([this, response = move(response)] {
  544. // If the server responded with "Connection: close", close the connection
  545. // as the server may or may not want to close the socket. Also, if this is
  546. // a legacy HTTP server (1.0 or older), assume close is the default value.
  547. if (auto result = response->headers().get("Connection"sv); result.has_value() ? result->equals_ignoring_ascii_case("close"sv) : m_legacy_connection)
  548. shutdown(ShutdownMode::CloseSocket);
  549. did_finish(response);
  550. });
  551. }
  552. }