1 /*
2 * nghttp2 - HTTP/2 C Library
3 *
4 * Copyright (c) 2012 Tatsuhiro Tsujikawa
5 *
6 * Permission is hereby granted, free of charge, to any person obtaining
7 * a copy of this software and associated documentation files (the
8 * "Software"), to deal in the Software without restriction, including
9 * without limitation the rights to use, copy, modify, merge, publish,
10 * distribute, sublicense, and/or sell copies of the Software, and to
11 * permit persons to whom the Software is furnished to do so, subject to
12 * the following conditions:
13 *
14 * The above copyright notice and this permission notice shall be
15 * included in all copies or substantial portions of the Software.
16 *
17 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
18 * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
19 * MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
20 * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
21 * LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
22 * OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
23 * WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
24 */
25 #include "shrpx_http_downstream_connection.h"
26
27 #include <openssl/rand.h>
28
29 #include "shrpx_client_handler.h"
30 #include "shrpx_upstream.h"
31 #include "shrpx_downstream.h"
32 #include "shrpx_config.h"
33 #include "shrpx_error.h"
34 #include "shrpx_http.h"
35 #include "shrpx_log_config.h"
36 #include "shrpx_connect_blocker.h"
37 #include "shrpx_downstream_connection_pool.h"
38 #include "shrpx_worker.h"
39 #include "shrpx_http2_session.h"
40 #include "shrpx_tls.h"
41 #include "shrpx_log.h"
42 #include "http2.h"
43 #include "util.h"
44 #include "ssl_compat.h"
45
46 using namespace nghttp2;
47
48 namespace shrpx {
49
50 namespace {
timeoutcb(struct ev_loop * loop,ev_timer * w,int revents)51 void timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
52 auto conn = static_cast<Connection *>(w->data);
53 auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
54
55 if (w == &conn->rt && !conn->expired_rt()) {
56 return;
57 }
58
59 if (LOG_ENABLED(INFO)) {
60 DCLOG(INFO, dconn) << "Time out";
61 }
62
63 auto downstream = dconn->get_downstream();
64 auto upstream = downstream->get_upstream();
65 auto handler = upstream->get_client_handler();
66 auto &resp = downstream->response();
67
68 // Do this so that dconn is not pooled
69 resp.connection_close = true;
70
71 if (upstream->downstream_error(dconn, Downstream::EVENT_TIMEOUT) != 0) {
72 delete handler;
73 }
74 }
75 } // namespace
76
77 namespace {
retry_downstream_connection(Downstream * downstream,unsigned int status_code)78 void retry_downstream_connection(Downstream *downstream,
79 unsigned int status_code) {
80 auto upstream = downstream->get_upstream();
81 auto handler = upstream->get_client_handler();
82
83 assert(!downstream->get_request_header_sent());
84
85 downstream->add_retry();
86
87 if (downstream->no_more_retry()) {
88 delete handler;
89 return;
90 }
91
92 downstream->pop_downstream_connection();
93 auto buf = downstream->get_request_buf();
94 buf->reset();
95
96 int rv;
97
98 for (;;) {
99 auto ndconn = handler->get_downstream_connection(rv, downstream);
100 if (!ndconn) {
101 break;
102 }
103 if (downstream->attach_downstream_connection(std::move(ndconn)) != 0) {
104 continue;
105 }
106 if (downstream->push_request_headers() == 0) {
107 return;
108 }
109 }
110
111 downstream->set_request_state(DownstreamState::CONNECT_FAIL);
112
113 if (rv == SHRPX_ERR_TLS_REQUIRED) {
114 rv = upstream->on_downstream_abort_request_with_https_redirect(downstream);
115 } else {
116 rv = upstream->on_downstream_abort_request(downstream, status_code);
117 }
118
119 if (rv != 0) {
120 delete handler;
121 }
122 }
123 } // namespace
124
125 namespace {
connect_timeoutcb(struct ev_loop * loop,ev_timer * w,int revents)126 void connect_timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
127 auto conn = static_cast<Connection *>(w->data);
128 auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
129 auto addr = dconn->get_addr();
130 auto raddr = dconn->get_raddr();
131
132 DCLOG(WARN, dconn) << "Connect time out; addr="
133 << util::to_numeric_addr(raddr);
134
135 downstream_failure(addr, raddr);
136
137 auto downstream = dconn->get_downstream();
138
139 retry_downstream_connection(downstream, 504);
140 }
141 } // namespace
142
143 namespace {
backend_retry(Downstream * downstream)144 void backend_retry(Downstream *downstream) {
145 retry_downstream_connection(downstream, 502);
146 }
147 } // namespace
148
149 namespace {
readcb(struct ev_loop * loop,ev_io * w,int revents)150 void readcb(struct ev_loop *loop, ev_io *w, int revents) {
151 int rv;
152 auto conn = static_cast<Connection *>(w->data);
153 auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
154 auto downstream = dconn->get_downstream();
155 auto upstream = downstream->get_upstream();
156 auto handler = upstream->get_client_handler();
157
158 rv = upstream->downstream_read(dconn);
159 if (rv != 0) {
160 if (rv == SHRPX_ERR_RETRY) {
161 backend_retry(downstream);
162 return;
163 }
164
165 delete handler;
166 }
167 }
168 } // namespace
169
170 namespace {
writecb(struct ev_loop * loop,ev_io * w,int revents)171 void writecb(struct ev_loop *loop, ev_io *w, int revents) {
172 int rv;
173 auto conn = static_cast<Connection *>(w->data);
174 auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
175 auto downstream = dconn->get_downstream();
176 auto upstream = downstream->get_upstream();
177 auto handler = upstream->get_client_handler();
178
179 rv = upstream->downstream_write(dconn);
180 if (rv == SHRPX_ERR_RETRY) {
181 backend_retry(downstream);
182 return;
183 }
184
185 if (rv != 0) {
186 delete handler;
187 }
188 }
189 } // namespace
190
191 namespace {
connectcb(struct ev_loop * loop,ev_io * w,int revents)192 void connectcb(struct ev_loop *loop, ev_io *w, int revents) {
193 auto conn = static_cast<Connection *>(w->data);
194 auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
195 auto downstream = dconn->get_downstream();
196 if (dconn->connected() != 0) {
197 backend_retry(downstream);
198 return;
199 }
200 writecb(loop, w, revents);
201 }
202 } // namespace
203
HttpDownstreamConnection(const std::shared_ptr<DownstreamAddrGroup> & group,DownstreamAddr * addr,struct ev_loop * loop,Worker * worker)204 HttpDownstreamConnection::HttpDownstreamConnection(
205 const std::shared_ptr<DownstreamAddrGroup> &group, DownstreamAddr *addr,
206 struct ev_loop *loop, Worker *worker)
207 : conn_(loop, -1, nullptr, worker->get_mcpool(),
208 group->shared_addr->timeout.write, group->shared_addr->timeout.read,
209 {}, {}, connectcb, readcb, connect_timeoutcb, this,
210 get_config()->tls.dyn_rec.warmup_threshold,
211 get_config()->tls.dyn_rec.idle_timeout, Proto::HTTP1),
212 on_read_(&HttpDownstreamConnection::noop),
213 on_write_(&HttpDownstreamConnection::noop),
214 signal_write_(&HttpDownstreamConnection::noop),
215 worker_(worker),
216 ssl_ctx_(worker->get_cl_ssl_ctx()),
217 group_(group),
218 addr_(addr),
219 raddr_(nullptr),
220 ioctrl_(&conn_.rlimit),
221 response_htp_{0},
222 first_write_done_(false),
223 reusable_(true),
224 request_header_written_(false) {}
225
~HttpDownstreamConnection()226 HttpDownstreamConnection::~HttpDownstreamConnection() {
227 if (LOG_ENABLED(INFO)) {
228 DCLOG(INFO, this) << "Deleted";
229 }
230
231 if (dns_query_) {
232 auto dns_tracker = worker_->get_dns_tracker();
233 dns_tracker->cancel(dns_query_.get());
234 }
235 }
236
attach_downstream(Downstream * downstream)237 int HttpDownstreamConnection::attach_downstream(Downstream *downstream) {
238 int rv;
239
240 if (LOG_ENABLED(INFO)) {
241 DCLOG(INFO, this) << "Attaching to DOWNSTREAM:" << downstream;
242 }
243
244 downstream_ = downstream;
245
246 rv = initiate_connection();
247 if (rv != 0) {
248 downstream_ = nullptr;
249 return rv;
250 }
251
252 return 0;
253 }
254
255 namespace {
256 int htp_msg_begincb(llhttp_t *htp);
257 int htp_hdr_keycb(llhttp_t *htp, const char *data, size_t len);
258 int htp_hdr_valcb(llhttp_t *htp, const char *data, size_t len);
259 int htp_hdrs_completecb(llhttp_t *htp);
260 int htp_bodycb(llhttp_t *htp, const char *data, size_t len);
261 int htp_msg_completecb(llhttp_t *htp);
262 } // namespace
263
264 namespace {
265 constexpr llhttp_settings_t htp_hooks = {
266 htp_msg_begincb, // llhttp_cb on_message_begin;
267 nullptr, // llhttp_data_cb on_url;
268 nullptr, // llhttp_data_cb on_status;
269 nullptr, // llhttp_data_cb on_method;
270 nullptr, // llhttp_data_cb on_version;
271 htp_hdr_keycb, // llhttp_data_cb on_header_field;
272 htp_hdr_valcb, // llhttp_data_cb on_header_value;
273 nullptr, // llhttp_data_cb on_chunk_extension_name;
274 nullptr, // llhttp_data_cb on_chunk_extension_value;
275 htp_hdrs_completecb, // llhttp_cb on_headers_complete;
276 htp_bodycb, // llhttp_data_cb on_body;
277 htp_msg_completecb, // llhttp_cb on_message_complete;
278 nullptr, // llhttp_cb on_url_complete;
279 nullptr, // llhttp_cb on_status_complete;
280 nullptr, // llhttp_cb on_method_complete;
281 nullptr, // llhttp_cb on_version_complete;
282 nullptr, // llhttp_cb on_header_field_complete;
283 nullptr, // llhttp_cb on_header_value_complete;
284 nullptr, // llhttp_cb on_chunk_extension_name_complete;
285 nullptr, // llhttp_cb on_chunk_extension_value_complete;
286 nullptr, // llhttp_cb on_chunk_header;
287 nullptr, // llhttp_cb on_chunk_complete;
288 nullptr, // llhttp_cb on_reset;
289 };
290 } // namespace
291
initiate_connection()292 int HttpDownstreamConnection::initiate_connection() {
293 int rv;
294
295 auto worker_blocker = worker_->get_connect_blocker();
296 if (worker_blocker->blocked()) {
297 if (LOG_ENABLED(INFO)) {
298 DCLOG(INFO, this)
299 << "Worker wide backend connection was blocked temporarily";
300 }
301 return SHRPX_ERR_NETWORK;
302 }
303
304 auto &downstreamconf = *worker_->get_downstream_config();
305
306 if (conn_.fd == -1) {
307 auto check_dns_result = dns_query_.get() != nullptr;
308
309 if (check_dns_result) {
310 assert(addr_->dns);
311 }
312
313 auto &connect_blocker = addr_->connect_blocker;
314
315 if (connect_blocker->blocked()) {
316 if (LOG_ENABLED(INFO)) {
317 DCLOG(INFO, this) << "Backend server " << addr_->host << ":"
318 << addr_->port << " was not available temporarily";
319 }
320
321 return SHRPX_ERR_NETWORK;
322 }
323
324 Address *raddr;
325
326 if (addr_->dns) {
327 if (!check_dns_result) {
328 auto dns_query = std::make_unique<DNSQuery>(
329 addr_->host,
330 [this](DNSResolverStatus status, const Address *result) {
331 int rv;
332
333 if (status == DNSResolverStatus::OK) {
334 *this->resolved_addr_ = *result;
335 }
336
337 rv = this->initiate_connection();
338 if (rv != 0) {
339 // This callback destroys |this|.
340 auto downstream = this->downstream_;
341 backend_retry(downstream);
342 }
343 });
344
345 auto dns_tracker = worker_->get_dns_tracker();
346
347 if (!resolved_addr_) {
348 resolved_addr_ = std::make_unique<Address>();
349 }
350 switch (dns_tracker->resolve(resolved_addr_.get(), dns_query.get())) {
351 case DNSResolverStatus::ERROR:
352 downstream_failure(addr_, nullptr);
353 return SHRPX_ERR_NETWORK;
354 case DNSResolverStatus::RUNNING:
355 dns_query_ = std::move(dns_query);
356 return 0;
357 case DNSResolverStatus::OK:
358 break;
359 default:
360 assert(0);
361 }
362 } else {
363 switch (dns_query_->status) {
364 case DNSResolverStatus::ERROR:
365 dns_query_.reset();
366 downstream_failure(addr_, nullptr);
367 return SHRPX_ERR_NETWORK;
368 case DNSResolverStatus::OK:
369 dns_query_.reset();
370 break;
371 default:
372 assert(0);
373 }
374 }
375
376 raddr = resolved_addr_.get();
377 util::set_port(*resolved_addr_, addr_->port);
378 } else {
379 raddr = &addr_->addr;
380 }
381
382 conn_.fd = util::create_nonblock_socket(raddr->su.storage.ss_family);
383
384 if (conn_.fd == -1) {
385 auto error = errno;
386 DCLOG(WARN, this) << "socket() failed; addr="
387 << util::to_numeric_addr(raddr) << ", errno=" << error;
388
389 worker_blocker->on_failure();
390
391 return SHRPX_ERR_NETWORK;
392 }
393
394 worker_blocker->on_success();
395
396 rv = connect(conn_.fd, &raddr->su.sa, raddr->len);
397 if (rv != 0 && errno != EINPROGRESS) {
398 auto error = errno;
399 DCLOG(WARN, this) << "connect() failed; addr="
400 << util::to_numeric_addr(raddr) << ", errno=" << error;
401
402 downstream_failure(addr_, raddr);
403
404 return SHRPX_ERR_NETWORK;
405 }
406
407 if (LOG_ENABLED(INFO)) {
408 DCLOG(INFO, this) << "Connecting to downstream server";
409 }
410
411 raddr_ = raddr;
412
413 if (addr_->tls) {
414 assert(ssl_ctx_);
415
416 auto ssl = tls::create_ssl(ssl_ctx_);
417 if (!ssl) {
418 return -1;
419 }
420
421 tls::setup_downstream_http1_alpn(ssl);
422
423 conn_.set_ssl(ssl);
424 conn_.tls.client_session_cache = &addr_->tls_session_cache;
425
426 auto sni_name =
427 addr_->sni.empty() ? StringRef{addr_->host} : StringRef{addr_->sni};
428 if (!util::numeric_host(sni_name.data())) {
429 SSL_set_tlsext_host_name(conn_.tls.ssl, sni_name.data());
430 }
431
432 auto session = tls::reuse_tls_session(addr_->tls_session_cache);
433 if (session) {
434 SSL_set_session(conn_.tls.ssl, session);
435 SSL_SESSION_free(session);
436 }
437
438 conn_.prepare_client_handshake();
439 }
440
441 ev_io_set(&conn_.wev, conn_.fd, EV_WRITE);
442 ev_io_set(&conn_.rev, conn_.fd, EV_READ);
443
444 conn_.wlimit.startw();
445
446 conn_.wt.repeat = downstreamconf.timeout.connect;
447 ev_timer_again(conn_.loop, &conn_.wt);
448 } else {
449 // we may set read timer cb to idle_timeoutcb. Reset again.
450 ev_set_cb(&conn_.rt, timeoutcb);
451 if (conn_.read_timeout < group_->shared_addr->timeout.read) {
452 conn_.read_timeout = group_->shared_addr->timeout.read;
453 conn_.last_read = std::chrono::steady_clock::now();
454 } else {
455 conn_.again_rt(group_->shared_addr->timeout.read);
456 }
457
458 ev_set_cb(&conn_.rev, readcb);
459
460 on_write_ = &HttpDownstreamConnection::write_first;
461 first_write_done_ = false;
462 request_header_written_ = false;
463 }
464
465 llhttp_init(&response_htp_, HTTP_RESPONSE, &htp_hooks);
466 response_htp_.data = downstream_;
467
468 return 0;
469 }
470
push_request_headers()471 int HttpDownstreamConnection::push_request_headers() {
472 if (request_header_written_) {
473 signal_write();
474 return 0;
475 }
476
477 const auto &downstream_hostport = addr_->hostport;
478 const auto &req = downstream_->request();
479
480 auto &balloc = downstream_->get_block_allocator();
481
482 auto connect_method = req.regular_connect_method();
483
484 auto config = get_config();
485 auto &httpconf = config->http;
486
487 request_header_written_ = true;
488
489 // For HTTP/1.0 request, there is no authority in request. In that
490 // case, we use backend server's host nonetheless.
491 auto authority = StringRef(downstream_hostport);
492 auto no_host_rewrite =
493 httpconf.no_host_rewrite || config->http2_proxy || connect_method;
494
495 if (no_host_rewrite && !req.authority.empty()) {
496 authority = req.authority;
497 }
498
499 downstream_->set_request_downstream_host(authority);
500
501 auto buf = downstream_->get_request_buf();
502
503 // Assume that method and request path do not contain \r\n.
504 auto meth = http2::to_method_string(
505 req.connect_proto == ConnectProto::WEBSOCKET ? HTTP_GET : req.method);
506 buf->append(meth);
507 buf->append(' ');
508
509 if (connect_method) {
510 buf->append(authority);
511 } else if (config->http2_proxy) {
512 // Construct absolute-form request target because we are going to
513 // send a request to a HTTP/1 proxy.
514 assert(!req.scheme.empty());
515 buf->append(req.scheme);
516 buf->append("://");
517 buf->append(authority);
518 buf->append(req.path);
519 } else if (req.method == HTTP_OPTIONS && req.path.empty()) {
520 // Server-wide OPTIONS
521 buf->append("*");
522 } else {
523 buf->append(req.path);
524 }
525 buf->append(" HTTP/1.1\r\nHost: ");
526 buf->append(authority);
527 buf->append("\r\n");
528
529 auto &fwdconf = httpconf.forwarded;
530 auto &xffconf = httpconf.xff;
531 auto &xfpconf = httpconf.xfp;
532 auto &earlydataconf = httpconf.early_data;
533
534 uint32_t build_flags =
535 (fwdconf.strip_incoming ? http2::HDOP_STRIP_FORWARDED : 0) |
536 (xffconf.strip_incoming ? http2::HDOP_STRIP_X_FORWARDED_FOR : 0) |
537 (xfpconf.strip_incoming ? http2::HDOP_STRIP_X_FORWARDED_PROTO : 0) |
538 (earlydataconf.strip_incoming ? http2::HDOP_STRIP_EARLY_DATA : 0) |
539 ((req.http_major == 3 || req.http_major == 2)
540 ? http2::HDOP_STRIP_SEC_WEBSOCKET_KEY
541 : 0);
542
543 http2::build_http1_headers_from_headers(buf, req.fs.headers(), build_flags);
544
545 auto cookie = downstream_->assemble_request_cookie();
546 if (!cookie.empty()) {
547 buf->append("Cookie: ");
548 buf->append(cookie);
549 buf->append("\r\n");
550 }
551
552 // set transfer-encoding only when content-length is unknown and
553 // request body is expected.
554 if (req.method != HTTP_CONNECT && req.http2_expect_body &&
555 req.fs.content_length == -1) {
556 downstream_->set_chunked_request(true);
557 buf->append("Transfer-Encoding: chunked\r\n");
558 }
559
560 if (req.connect_proto == ConnectProto::WEBSOCKET) {
561 if (req.http_major == 3 || req.http_major == 2) {
562 std::array<uint8_t, 16> nonce;
563 if (RAND_bytes(nonce.data(), nonce.size()) != 1) {
564 return -1;
565 }
566 auto iov = make_byte_ref(balloc, base64::encode_length(nonce.size()) + 1);
567 auto p =
568 base64::encode(std::begin(nonce), std::end(nonce), std::begin(iov));
569 *p = '\0';
570 auto key = StringRef{std::span{std::begin(iov), p}};
571 downstream_->set_ws_key(key);
572
573 buf->append("Sec-Websocket-Key: ");
574 buf->append(key);
575 buf->append("\r\n");
576 }
577
578 buf->append("Upgrade: websocket\r\nConnection: Upgrade\r\n");
579 } else if (!connect_method && req.upgrade_request) {
580 auto connection = req.fs.header(http2::HD_CONNECTION);
581 if (connection) {
582 buf->append("Connection: ");
583 buf->append((*connection).value);
584 buf->append("\r\n");
585 }
586
587 auto upgrade = req.fs.header(http2::HD_UPGRADE);
588 if (upgrade) {
589 buf->append("Upgrade: ");
590 buf->append((*upgrade).value);
591 buf->append("\r\n");
592 }
593 } else if (req.connection_close) {
594 buf->append("Connection: close\r\n");
595 }
596
597 auto upstream = downstream_->get_upstream();
598 auto handler = upstream->get_client_handler();
599
600 #if defined(NGHTTP2_GENUINE_OPENSSL) || defined(NGHTTP2_OPENSSL_IS_BORINGSSL)
601 auto conn = handler->get_connection();
602
603 if (conn->tls.ssl && !SSL_is_init_finished(conn->tls.ssl)) {
604 buf->append("Early-Data: 1\r\n");
605 }
606 #endif // NGHTTP2_GENUINE_OPENSSL || NGHTTP2_OPENSSL_IS_BORINGSSL
607
608 auto fwd =
609 fwdconf.strip_incoming ? nullptr : req.fs.header(http2::HD_FORWARDED);
610
611 if (fwdconf.params) {
612 auto params = fwdconf.params;
613
614 if (config->http2_proxy || connect_method) {
615 params &= ~FORWARDED_PROTO;
616 }
617
618 auto value = http::create_forwarded(
619 balloc, params, handler->get_forwarded_by(),
620 handler->get_forwarded_for(), req.authority, req.scheme);
621
622 if (fwd || !value.empty()) {
623 buf->append("Forwarded: ");
624 if (fwd) {
625 buf->append(fwd->value);
626
627 if (!value.empty()) {
628 buf->append(", ");
629 }
630 }
631 buf->append(value);
632 buf->append("\r\n");
633 }
634 } else if (fwd) {
635 buf->append("Forwarded: ");
636 buf->append(fwd->value);
637 buf->append("\r\n");
638 }
639
640 auto xff = xffconf.strip_incoming ? nullptr
641 : req.fs.header(http2::HD_X_FORWARDED_FOR);
642
643 if (xffconf.add) {
644 buf->append("X-Forwarded-For: ");
645 if (xff) {
646 buf->append((*xff).value);
647 buf->append(", ");
648 }
649 buf->append(client_handler_->get_ipaddr());
650 buf->append("\r\n");
651 } else if (xff) {
652 buf->append("X-Forwarded-For: ");
653 buf->append((*xff).value);
654 buf->append("\r\n");
655 }
656 if (!config->http2_proxy && !connect_method) {
657 auto xfp = xfpconf.strip_incoming
658 ? nullptr
659 : req.fs.header(http2::HD_X_FORWARDED_PROTO);
660
661 if (xfpconf.add) {
662 buf->append("X-Forwarded-Proto: ");
663 if (xfp) {
664 buf->append((*xfp).value);
665 buf->append(", ");
666 }
667 assert(!req.scheme.empty());
668 buf->append(req.scheme);
669 buf->append("\r\n");
670 } else if (xfp) {
671 buf->append("X-Forwarded-Proto: ");
672 buf->append((*xfp).value);
673 buf->append("\r\n");
674 }
675 }
676 auto via = req.fs.header(http2::HD_VIA);
677 if (httpconf.no_via) {
678 if (via) {
679 buf->append("Via: ");
680 buf->append((*via).value);
681 buf->append("\r\n");
682 }
683 } else {
684 buf->append("Via: ");
685 if (via) {
686 buf->append((*via).value);
687 buf->append(", ");
688 }
689 std::array<char, 16> viabuf;
690 auto end = http::create_via_header_value(viabuf.data(), req.http_major,
691 req.http_minor);
692 buf->append(viabuf.data(), end - viabuf.data());
693 buf->append("\r\n");
694 }
695
696 for (auto &p : httpconf.add_request_headers) {
697 buf->append(p.name);
698 buf->append(": ");
699 buf->append(p.value);
700 buf->append("\r\n");
701 }
702
703 buf->append("\r\n");
704
705 if (LOG_ENABLED(INFO)) {
706 std::string nhdrs;
707 for (auto chunk = buf->head; chunk; chunk = chunk->next) {
708 nhdrs.append(chunk->pos, chunk->last);
709 }
710 if (log_config()->errorlog_tty) {
711 nhdrs = http::colorizeHeaders(nhdrs.c_str());
712 }
713 DCLOG(INFO, this) << "HTTP request headers. stream_id="
714 << downstream_->get_stream_id() << "\n"
715 << nhdrs;
716 }
717
718 // Don't call signal_write() if we anticipate request body. We call
719 // signal_write() when we received request body chunk, and it
720 // enables us to send headers and data in one writev system call.
721 if (req.method == HTTP_CONNECT ||
722 downstream_->get_blocked_request_buf()->rleft() ||
723 (!req.http2_expect_body && req.fs.content_length == 0) ||
724 downstream_->get_expect_100_continue()) {
725 signal_write();
726 }
727
728 return 0;
729 }
730
process_blocked_request_buf()731 int HttpDownstreamConnection::process_blocked_request_buf() {
732 auto src = downstream_->get_blocked_request_buf();
733
734 if (src->rleft()) {
735 auto dest = downstream_->get_request_buf();
736 auto chunked = downstream_->get_chunked_request();
737 if (chunked) {
738 auto chunk_size_hex = util::utox(src->rleft());
739 dest->append(chunk_size_hex);
740 dest->append("\r\n");
741 }
742
743 src->copy(*dest);
744
745 if (chunked) {
746 dest->append("\r\n");
747 }
748 }
749
750 if (downstream_->get_blocked_request_data_eof() &&
751 downstream_->get_chunked_request()) {
752 end_upload_data_chunk();
753 }
754
755 return 0;
756 }
757
push_upload_data_chunk(const uint8_t * data,size_t datalen)758 int HttpDownstreamConnection::push_upload_data_chunk(const uint8_t *data,
759 size_t datalen) {
760 if (!downstream_->get_request_header_sent()) {
761 auto output = downstream_->get_blocked_request_buf();
762 auto &req = downstream_->request();
763 output->append(data, datalen);
764 req.unconsumed_body_length += datalen;
765 if (request_header_written_) {
766 signal_write();
767 }
768 return 0;
769 }
770
771 auto chunked = downstream_->get_chunked_request();
772 auto output = downstream_->get_request_buf();
773
774 if (chunked) {
775 auto chunk_size_hex = util::utox(datalen);
776 output->append(chunk_size_hex);
777 output->append("\r\n");
778 }
779
780 output->append(data, datalen);
781
782 if (chunked) {
783 output->append("\r\n");
784 }
785
786 signal_write();
787
788 return 0;
789 }
790
end_upload_data()791 int HttpDownstreamConnection::end_upload_data() {
792 if (!downstream_->get_request_header_sent()) {
793 downstream_->set_blocked_request_data_eof(true);
794 if (request_header_written_) {
795 signal_write();
796 }
797 return 0;
798 }
799
800 signal_write();
801
802 if (!downstream_->get_chunked_request()) {
803 return 0;
804 }
805
806 end_upload_data_chunk();
807
808 return 0;
809 }
810
end_upload_data_chunk()811 void HttpDownstreamConnection::end_upload_data_chunk() {
812 const auto &req = downstream_->request();
813
814 auto output = downstream_->get_request_buf();
815 const auto &trailers = req.fs.trailers();
816 if (trailers.empty()) {
817 output->append("0\r\n\r\n");
818 } else {
819 output->append("0\r\n");
820 http2::build_http1_headers_from_headers(output, trailers,
821 http2::HDOP_STRIP_ALL);
822 output->append("\r\n");
823 }
824 }
825
826 namespace {
remove_from_pool(HttpDownstreamConnection * dconn)827 void remove_from_pool(HttpDownstreamConnection *dconn) {
828 auto addr = dconn->get_addr();
829 auto &dconn_pool = addr->dconn_pool;
830 dconn_pool->remove_downstream_connection(dconn);
831 }
832 } // namespace
833
834 namespace {
idle_readcb(struct ev_loop * loop,ev_io * w,int revents)835 void idle_readcb(struct ev_loop *loop, ev_io *w, int revents) {
836 auto conn = static_cast<Connection *>(w->data);
837 auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
838 if (LOG_ENABLED(INFO)) {
839 DCLOG(INFO, dconn) << "Idle connection EOF";
840 }
841
842 remove_from_pool(dconn);
843 // dconn was deleted
844 }
845 } // namespace
846
847 namespace {
idle_timeoutcb(struct ev_loop * loop,ev_timer * w,int revents)848 void idle_timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
849 auto conn = static_cast<Connection *>(w->data);
850 auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
851
852 if (w == &conn->rt && !conn->expired_rt()) {
853 return;
854 }
855
856 if (LOG_ENABLED(INFO)) {
857 DCLOG(INFO, dconn) << "Idle connection timeout";
858 }
859
860 remove_from_pool(dconn);
861 // dconn was deleted
862 }
863 } // namespace
864
detach_downstream(Downstream * downstream)865 void HttpDownstreamConnection::detach_downstream(Downstream *downstream) {
866 if (LOG_ENABLED(INFO)) {
867 DCLOG(INFO, this) << "Detaching from DOWNSTREAM:" << downstream;
868 }
869 downstream_ = nullptr;
870
871 ev_set_cb(&conn_.rev, idle_readcb);
872 ioctrl_.force_resume_read();
873
874 auto &downstreamconf = *worker_->get_downstream_config();
875
876 ev_set_cb(&conn_.rt, idle_timeoutcb);
877 if (conn_.read_timeout < downstreamconf.timeout.idle_read) {
878 conn_.read_timeout = downstreamconf.timeout.idle_read;
879 conn_.last_read = std::chrono::steady_clock::now();
880 } else {
881 conn_.again_rt(downstreamconf.timeout.idle_read);
882 }
883
884 conn_.wlimit.stopw();
885 ev_timer_stop(conn_.loop, &conn_.wt);
886 }
887
pause_read(IOCtrlReason reason)888 void HttpDownstreamConnection::pause_read(IOCtrlReason reason) {
889 ioctrl_.pause_read(reason);
890 }
891
resume_read(IOCtrlReason reason,size_t consumed)892 int HttpDownstreamConnection::resume_read(IOCtrlReason reason,
893 size_t consumed) {
894 auto &downstreamconf = *worker_->get_downstream_config();
895
896 if (downstream_->get_response_buf()->rleft() <=
897 downstreamconf.request_buffer_size / 2) {
898 ioctrl_.resume_read(reason);
899 }
900
901 return 0;
902 }
903
force_resume_read()904 void HttpDownstreamConnection::force_resume_read() {
905 ioctrl_.force_resume_read();
906 }
907
908 namespace {
htp_msg_begincb(llhttp_t * htp)909 int htp_msg_begincb(llhttp_t *htp) {
910 auto downstream = static_cast<Downstream *>(htp->data);
911
912 if (downstream->get_response_state() != DownstreamState::INITIAL) {
913 return -1;
914 }
915
916 return 0;
917 }
918 } // namespace
919
920 namespace {
htp_hdrs_completecb(llhttp_t * htp)921 int htp_hdrs_completecb(llhttp_t *htp) {
922 auto downstream = static_cast<Downstream *>(htp->data);
923 auto upstream = downstream->get_upstream();
924 auto handler = upstream->get_client_handler();
925 const auto &req = downstream->request();
926 auto &resp = downstream->response();
927 int rv;
928
929 auto &balloc = downstream->get_block_allocator();
930
931 for (auto &kv : resp.fs.headers()) {
932 kv.value = util::rstrip(balloc, kv.value);
933
934 if (kv.token == http2::HD_TRANSFER_ENCODING &&
935 !http2::check_transfer_encoding(kv.value)) {
936 return -1;
937 }
938 }
939
940 auto config = get_config();
941 auto &loggingconf = config->logging;
942
943 resp.http_status = htp->status_code;
944 resp.http_major = htp->http_major;
945 resp.http_minor = htp->http_minor;
946
947 if (resp.http_major > 1 || req.http_minor > 1) {
948 resp.http_major = 1;
949 resp.http_minor = 1;
950 return -1;
951 }
952
953 auto dconn = downstream->get_downstream_connection();
954
955 downstream->set_downstream_addr_group(dconn->get_downstream_addr_group());
956 downstream->set_addr(dconn->get_addr());
957
958 // Server MUST NOT send Transfer-Encoding with a status code 1xx or
959 // 204. Also server MUST NOT send Transfer-Encoding with a status
960 // code 2xx to a CONNECT request. Same holds true with
961 // Content-Length.
962 if (resp.http_status == 204) {
963 if (resp.fs.header(http2::HD_TRANSFER_ENCODING)) {
964 return -1;
965 }
966 // Some server send content-length: 0 for 204. Until they get
967 // fixed, we accept, but ignore it.
968
969 // Calling parse_content_length() detects duplicated
970 // content-length header fields.
971 if (resp.fs.parse_content_length() != 0) {
972 return -1;
973 }
974 if (resp.fs.content_length == 0) {
975 resp.fs.erase_content_length_and_transfer_encoding();
976 } else if (resp.fs.content_length != -1) {
977 return -1;
978 }
979 } else if (resp.http_status / 100 == 1 ||
980 (resp.http_status / 100 == 2 && req.method == HTTP_CONNECT)) {
981 // Server MUST NOT send Content-Length and Transfer-Encoding in
982 // these responses.
983 resp.fs.erase_content_length_and_transfer_encoding();
984 } else if (resp.fs.parse_content_length() != 0) {
985 downstream->set_response_state(DownstreamState::MSG_BAD_HEADER);
986 return -1;
987 }
988
989 // Check upgrade before processing non-final response, since if
990 // upgrade succeeded, 101 response is treated as final in nghttpx.
991 downstream->check_upgrade_fulfilled_http1();
992
993 if (downstream->get_non_final_response()) {
994 // Reset content-length because we reuse same Downstream for the
995 // next response.
996 resp.fs.content_length = -1;
997 // For non-final response code, we just call
998 // on_downstream_header_complete() without changing response
999 // state.
1000 rv = upstream->on_downstream_header_complete(downstream);
1001
1002 if (rv != 0) {
1003 return -1;
1004 }
1005
1006 // Ignore response body for non-final response.
1007 return 1;
1008 }
1009
1010 resp.connection_close = !llhttp_should_keep_alive(htp);
1011 downstream->set_response_state(DownstreamState::HEADER_COMPLETE);
1012 downstream->inspect_http1_response();
1013
1014 if (htp->flags & F_CHUNKED) {
1015 downstream->set_chunked_response(true);
1016 }
1017
1018 auto transfer_encoding = resp.fs.header(http2::HD_TRANSFER_ENCODING);
1019 if (transfer_encoding && !downstream->get_chunked_response()) {
1020 resp.connection_close = true;
1021 }
1022
1023 if (downstream->get_upgraded()) {
1024 // content-length must be ignored for upgraded connection.
1025 resp.fs.content_length = -1;
1026 resp.connection_close = true;
1027 // transfer-encoding not applied to upgraded connection
1028 downstream->set_chunked_response(false);
1029 } else if (http2::legacy_http1(req.http_major, req.http_minor)) {
1030 if (resp.fs.content_length == -1) {
1031 resp.connection_close = true;
1032 }
1033 downstream->set_chunked_response(false);
1034 } else if (!downstream->expect_response_body()) {
1035 downstream->set_chunked_response(false);
1036 }
1037
1038 if (loggingconf.access.write_early && downstream->accesslog_ready()) {
1039 handler->write_accesslog(downstream);
1040 downstream->set_accesslog_written(true);
1041 }
1042
1043 if (upstream->on_downstream_header_complete(downstream) != 0) {
1044 return -1;
1045 }
1046
1047 if (downstream->get_upgraded()) {
1048 // Upgrade complete, read until EOF in both ends
1049 if (upstream->resume_read(SHRPX_NO_BUFFER, downstream, 0) != 0) {
1050 return -1;
1051 }
1052 downstream->set_request_state(DownstreamState::HEADER_COMPLETE);
1053 if (LOG_ENABLED(INFO)) {
1054 LOG(INFO) << "HTTP upgrade success. stream_id="
1055 << downstream->get_stream_id();
1056 }
1057 }
1058
1059 // Ignore the response body. HEAD response may contain
1060 // Content-Length or Transfer-Encoding: chunked. Some server send
1061 // 304 status code with nonzero Content-Length, but without response
1062 // body. See
1063 // https://tools.ietf.org/html/rfc7230#section-3.3
1064
1065 // TODO It seems that the cases other than HEAD are handled by
1066 // llhttp. Need test.
1067 return !http2::expect_response_body(req.method, resp.http_status);
1068 }
1069 } // namespace
1070
1071 namespace {
ensure_header_field_buffer(const Downstream * downstream,const HttpConfig & httpconf,size_t len)1072 int ensure_header_field_buffer(const Downstream *downstream,
1073 const HttpConfig &httpconf, size_t len) {
1074 auto &resp = downstream->response();
1075
1076 if (resp.fs.buffer_size() + len > httpconf.response_header_field_buffer) {
1077 if (LOG_ENABLED(INFO)) {
1078 DLOG(INFO, downstream) << "Too large header header field size="
1079 << resp.fs.buffer_size() + len;
1080 }
1081 return -1;
1082 }
1083
1084 return 0;
1085 }
1086 } // namespace
1087
1088 namespace {
ensure_max_header_fields(const Downstream * downstream,const HttpConfig & httpconf)1089 int ensure_max_header_fields(const Downstream *downstream,
1090 const HttpConfig &httpconf) {
1091 auto &resp = downstream->response();
1092
1093 if (resp.fs.num_fields() >= httpconf.max_response_header_fields) {
1094 if (LOG_ENABLED(INFO)) {
1095 DLOG(INFO, downstream)
1096 << "Too many header field num=" << resp.fs.num_fields() + 1;
1097 }
1098 return -1;
1099 }
1100
1101 return 0;
1102 }
1103 } // namespace
1104
1105 namespace {
htp_hdr_keycb(llhttp_t * htp,const char * data,size_t len)1106 int htp_hdr_keycb(llhttp_t *htp, const char *data, size_t len) {
1107 auto downstream = static_cast<Downstream *>(htp->data);
1108 auto &resp = downstream->response();
1109 auto &httpconf = get_config()->http;
1110
1111 if (ensure_header_field_buffer(downstream, httpconf, len) != 0) {
1112 return -1;
1113 }
1114
1115 if (downstream->get_response_state() == DownstreamState::INITIAL) {
1116 if (resp.fs.header_key_prev()) {
1117 resp.fs.append_last_header_key(data, len);
1118 } else {
1119 if (ensure_max_header_fields(downstream, httpconf) != 0) {
1120 return -1;
1121 }
1122 resp.fs.alloc_add_header_name(StringRef{data, len});
1123 }
1124 } else {
1125 // trailer part
1126 if (resp.fs.trailer_key_prev()) {
1127 resp.fs.append_last_trailer_key(data, len);
1128 } else {
1129 if (ensure_max_header_fields(downstream, httpconf) != 0) {
1130 // Could not ignore this trailer field easily, since we may
1131 // get its value in htp_hdr_valcb, and it will be added to
1132 // wrong place or crash if trailer fields are currently empty.
1133 return -1;
1134 }
1135 resp.fs.alloc_add_trailer_name(StringRef{data, len});
1136 }
1137 }
1138 return 0;
1139 }
1140 } // namespace
1141
1142 namespace {
htp_hdr_valcb(llhttp_t * htp,const char * data,size_t len)1143 int htp_hdr_valcb(llhttp_t *htp, const char *data, size_t len) {
1144 auto downstream = static_cast<Downstream *>(htp->data);
1145 auto &resp = downstream->response();
1146 auto &httpconf = get_config()->http;
1147
1148 if (ensure_header_field_buffer(downstream, httpconf, len) != 0) {
1149 return -1;
1150 }
1151
1152 if (downstream->get_response_state() == DownstreamState::INITIAL) {
1153 resp.fs.append_last_header_value(data, len);
1154 } else {
1155 resp.fs.append_last_trailer_value(data, len);
1156 }
1157 return 0;
1158 }
1159 } // namespace
1160
1161 namespace {
htp_bodycb(llhttp_t * htp,const char * data,size_t len)1162 int htp_bodycb(llhttp_t *htp, const char *data, size_t len) {
1163 auto downstream = static_cast<Downstream *>(htp->data);
1164 auto &resp = downstream->response();
1165
1166 resp.recv_body_length += len;
1167
1168 return downstream->get_upstream()->on_downstream_body(
1169 downstream, reinterpret_cast<const uint8_t *>(data), len, true);
1170 }
1171 } // namespace
1172
1173 namespace {
htp_msg_completecb(llhttp_t * htp)1174 int htp_msg_completecb(llhttp_t *htp) {
1175 auto downstream = static_cast<Downstream *>(htp->data);
1176 auto &resp = downstream->response();
1177 auto &balloc = downstream->get_block_allocator();
1178
1179 for (auto &kv : resp.fs.trailers()) {
1180 kv.value = util::rstrip(balloc, kv.value);
1181 }
1182
1183 // llhttp does not treat "200 connection established" response
1184 // against CONNECT request, and in that case, this function is not
1185 // called. But if HTTP Upgrade is made (e.g., WebSocket), this
1186 // function is called, and llhttp_execute() returns just after that.
1187 if (downstream->get_upgraded()) {
1188 return 0;
1189 }
1190
1191 if (downstream->get_non_final_response()) {
1192 downstream->reset_response();
1193
1194 return 0;
1195 }
1196
1197 downstream->set_response_state(DownstreamState::MSG_COMPLETE);
1198 // Block reading another response message from (broken?)
1199 // server. This callback is not called if the connection is
1200 // tunneled.
1201 downstream->pause_read(SHRPX_MSG_BLOCK);
1202 return downstream->get_upstream()->on_downstream_body_complete(downstream);
1203 }
1204 } // namespace
1205
write_first()1206 int HttpDownstreamConnection::write_first() {
1207 int rv;
1208
1209 process_blocked_request_buf();
1210
1211 if (conn_.tls.ssl) {
1212 rv = write_tls();
1213 } else {
1214 rv = write_clear();
1215 }
1216
1217 if (rv != 0) {
1218 return SHRPX_ERR_RETRY;
1219 }
1220
1221 if (conn_.tls.ssl) {
1222 on_write_ = &HttpDownstreamConnection::write_tls;
1223 } else {
1224 on_write_ = &HttpDownstreamConnection::write_clear;
1225 }
1226
1227 first_write_done_ = true;
1228 downstream_->set_request_header_sent(true);
1229
1230 auto buf = downstream_->get_blocked_request_buf();
1231 buf->reset();
1232
1233 // upstream->resume_read() might be called in
1234 // write_tls()/write_clear(), but before blocked_request_buf_ is
1235 // reset. So upstream read might still be blocked. Let's do it
1236 // again here.
1237 auto input = downstream_->get_request_buf();
1238 if (input->rleft() == 0) {
1239 auto upstream = downstream_->get_upstream();
1240 auto &req = downstream_->request();
1241
1242 upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
1243 req.unconsumed_body_length);
1244 }
1245
1246 return 0;
1247 }
1248
read_clear()1249 int HttpDownstreamConnection::read_clear() {
1250 conn_.last_read = std::chrono::steady_clock::now();
1251
1252 std::array<uint8_t, 16_k> buf;
1253 int rv;
1254
1255 for (;;) {
1256 auto nread = conn_.read_clear(buf.data(), buf.size());
1257 if (nread == 0) {
1258 return 0;
1259 }
1260
1261 if (nread < 0) {
1262 if (nread == SHRPX_ERR_EOF && !downstream_->get_upgraded()) {
1263 auto htperr = llhttp_finish(&response_htp_);
1264 if (htperr != HPE_OK) {
1265 if (LOG_ENABLED(INFO)) {
1266 DCLOG(INFO, this) << "HTTP response ended prematurely: "
1267 << llhttp_errno_name(htperr);
1268 }
1269
1270 return -1;
1271 }
1272 }
1273
1274 return nread;
1275 }
1276
1277 rv = process_input(buf.data(), nread);
1278 if (rv != 0) {
1279 return rv;
1280 }
1281
1282 if (!ev_is_active(&conn_.rev)) {
1283 return 0;
1284 }
1285 }
1286 }
1287
write_clear()1288 int HttpDownstreamConnection::write_clear() {
1289 conn_.last_read = std::chrono::steady_clock::now();
1290
1291 auto upstream = downstream_->get_upstream();
1292 auto input = downstream_->get_request_buf();
1293
1294 std::array<struct iovec, MAX_WR_IOVCNT> iov;
1295
1296 while (input->rleft() > 0) {
1297 auto iovcnt = input->riovec(iov.data(), iov.size());
1298
1299 auto nwrite = conn_.writev_clear(iov.data(), iovcnt);
1300
1301 if (nwrite == 0) {
1302 return 0;
1303 }
1304
1305 if (nwrite < 0) {
1306 if (!first_write_done_) {
1307 return nwrite;
1308 }
1309 // We may have pending data in receive buffer which may contain
1310 // part of response body. So keep reading. Invoke read event
1311 // to get read(2) error just in case.
1312 ev_feed_event(conn_.loop, &conn_.rev, EV_READ);
1313 on_write_ = &HttpDownstreamConnection::noop;
1314 reusable_ = false;
1315 break;
1316 }
1317
1318 input->drain(nwrite);
1319 }
1320
1321 conn_.wlimit.stopw();
1322 ev_timer_stop(conn_.loop, &conn_.wt);
1323
1324 if (input->rleft() == 0) {
1325 auto &req = downstream_->request();
1326
1327 upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
1328 req.unconsumed_body_length);
1329 }
1330
1331 return 0;
1332 }
1333
tls_handshake()1334 int HttpDownstreamConnection::tls_handshake() {
1335 ERR_clear_error();
1336
1337 conn_.last_read = std::chrono::steady_clock::now();
1338
1339 auto rv = conn_.tls_handshake();
1340 if (rv == SHRPX_ERR_INPROGRESS) {
1341 return 0;
1342 }
1343
1344 if (rv < 0) {
1345 downstream_failure(addr_, raddr_);
1346
1347 return rv;
1348 }
1349
1350 if (LOG_ENABLED(INFO)) {
1351 DCLOG(INFO, this) << "SSL/TLS handshake completed";
1352 }
1353
1354 if (!get_config()->tls.insecure &&
1355 tls::check_cert(conn_.tls.ssl, addr_, raddr_) != 0) {
1356 downstream_failure(addr_, raddr_);
1357
1358 return -1;
1359 }
1360
1361 auto &connect_blocker = addr_->connect_blocker;
1362
1363 signal_write_ = &HttpDownstreamConnection::actual_signal_write;
1364
1365 connect_blocker->on_success();
1366
1367 ev_set_cb(&conn_.rt, timeoutcb);
1368 ev_set_cb(&conn_.wt, timeoutcb);
1369
1370 on_read_ = &HttpDownstreamConnection::read_tls;
1371 on_write_ = &HttpDownstreamConnection::write_first;
1372
1373 // TODO Check negotiated ALPN
1374
1375 return on_write();
1376 }
1377
read_tls()1378 int HttpDownstreamConnection::read_tls() {
1379 conn_.last_read = std::chrono::steady_clock::now();
1380
1381 ERR_clear_error();
1382
1383 std::array<uint8_t, 16_k> buf;
1384 int rv;
1385
1386 for (;;) {
1387 auto nread = conn_.read_tls(buf.data(), buf.size());
1388 if (nread == 0) {
1389 return 0;
1390 }
1391
1392 if (nread < 0) {
1393 if (nread == SHRPX_ERR_EOF && !downstream_->get_upgraded()) {
1394 auto htperr = llhttp_finish(&response_htp_);
1395 if (htperr != HPE_OK) {
1396 if (LOG_ENABLED(INFO)) {
1397 DCLOG(INFO, this) << "HTTP response ended prematurely: "
1398 << llhttp_errno_name(htperr);
1399 }
1400
1401 return -1;
1402 }
1403 }
1404
1405 return nread;
1406 }
1407
1408 rv = process_input(buf.data(), nread);
1409 if (rv != 0) {
1410 return rv;
1411 }
1412
1413 if (!ev_is_active(&conn_.rev)) {
1414 return 0;
1415 }
1416 }
1417 }
1418
write_tls()1419 int HttpDownstreamConnection::write_tls() {
1420 conn_.last_read = std::chrono::steady_clock::now();
1421
1422 ERR_clear_error();
1423
1424 auto upstream = downstream_->get_upstream();
1425 auto input = downstream_->get_request_buf();
1426
1427 struct iovec iov;
1428
1429 while (input->rleft() > 0) {
1430 auto iovcnt = input->riovec(&iov, 1);
1431 if (iovcnt != 1) {
1432 assert(0);
1433 return -1;
1434 }
1435 auto nwrite = conn_.write_tls(iov.iov_base, iov.iov_len);
1436
1437 if (nwrite == 0) {
1438 return 0;
1439 }
1440
1441 if (nwrite < 0) {
1442 if (!first_write_done_) {
1443 return nwrite;
1444 }
1445 // We may have pending data in receive buffer which may contain
1446 // part of response body. So keep reading. Invoke read event
1447 // to get read(2) error just in case.
1448 ev_feed_event(conn_.loop, &conn_.rev, EV_READ);
1449 on_write_ = &HttpDownstreamConnection::noop;
1450 reusable_ = false;
1451 break;
1452 }
1453
1454 input->drain(nwrite);
1455 }
1456
1457 conn_.wlimit.stopw();
1458 ev_timer_stop(conn_.loop, &conn_.wt);
1459
1460 if (input->rleft() == 0) {
1461 auto &req = downstream_->request();
1462
1463 upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
1464 req.unconsumed_body_length);
1465 }
1466
1467 return 0;
1468 }
1469
process_input(const uint8_t * data,size_t datalen)1470 int HttpDownstreamConnection::process_input(const uint8_t *data,
1471 size_t datalen) {
1472 int rv;
1473
1474 if (downstream_->get_upgraded()) {
1475 // For upgraded connection, just pass data to the upstream.
1476 rv = downstream_->get_upstream()->on_downstream_body(downstream_, data,
1477 datalen, true);
1478 if (rv != 0) {
1479 return rv;
1480 }
1481
1482 if (downstream_->response_buf_full()) {
1483 downstream_->pause_read(SHRPX_NO_BUFFER);
1484 return 0;
1485 }
1486
1487 return 0;
1488 }
1489
1490 auto htperr = llhttp_execute(&response_htp_,
1491 reinterpret_cast<const char *>(data), datalen);
1492 auto nproc =
1493 htperr == HPE_OK
1494 ? datalen
1495 : static_cast<size_t>(reinterpret_cast<const uint8_t *>(
1496 llhttp_get_error_pos(&response_htp_)) -
1497 data);
1498
1499 if (htperr != HPE_OK &&
1500 (!downstream_->get_upgraded() || htperr != HPE_PAUSED_UPGRADE)) {
1501 // Handling early return (in other words, response was hijacked by
1502 // mruby scripting).
1503 if (downstream_->get_response_state() == DownstreamState::MSG_COMPLETE) {
1504 return SHRPX_ERR_DCONN_CANCELED;
1505 }
1506
1507 if (LOG_ENABLED(INFO)) {
1508 DCLOG(INFO, this) << "HTTP parser failure: "
1509 << "(" << llhttp_errno_name(htperr) << ") "
1510 << llhttp_get_error_reason(&response_htp_);
1511 }
1512
1513 return -1;
1514 }
1515
1516 if (downstream_->get_upgraded()) {
1517 if (nproc < datalen) {
1518 // Data from data + nproc are for upgraded protocol.
1519 rv = downstream_->get_upstream()->on_downstream_body(
1520 downstream_, data + nproc, datalen - nproc, true);
1521 if (rv != 0) {
1522 return rv;
1523 }
1524
1525 if (downstream_->response_buf_full()) {
1526 downstream_->pause_read(SHRPX_NO_BUFFER);
1527 return 0;
1528 }
1529 }
1530 return 0;
1531 }
1532
1533 if (downstream_->response_buf_full()) {
1534 downstream_->pause_read(SHRPX_NO_BUFFER);
1535 return 0;
1536 }
1537
1538 return 0;
1539 }
1540
connected()1541 int HttpDownstreamConnection::connected() {
1542 auto &connect_blocker = addr_->connect_blocker;
1543
1544 auto sock_error = util::get_socket_error(conn_.fd);
1545 if (sock_error != 0) {
1546 conn_.wlimit.stopw();
1547
1548 DCLOG(WARN, this) << "Backend connect failed; addr="
1549 << util::to_numeric_addr(raddr_)
1550 << ": errno=" << sock_error;
1551
1552 downstream_failure(addr_, raddr_);
1553
1554 return -1;
1555 }
1556
1557 if (LOG_ENABLED(INFO)) {
1558 DCLOG(INFO, this) << "Connected to downstream host";
1559 }
1560
1561 // Reset timeout for write. Previously, we set timeout for connect.
1562 conn_.wt.repeat = group_->shared_addr->timeout.write;
1563 ev_timer_again(conn_.loop, &conn_.wt);
1564
1565 conn_.rlimit.startw();
1566 conn_.again_rt();
1567
1568 ev_set_cb(&conn_.wev, writecb);
1569
1570 if (conn_.tls.ssl) {
1571 on_read_ = &HttpDownstreamConnection::tls_handshake;
1572 on_write_ = &HttpDownstreamConnection::tls_handshake;
1573
1574 return 0;
1575 }
1576
1577 signal_write_ = &HttpDownstreamConnection::actual_signal_write;
1578
1579 connect_blocker->on_success();
1580
1581 ev_set_cb(&conn_.rt, timeoutcb);
1582 ev_set_cb(&conn_.wt, timeoutcb);
1583
1584 on_read_ = &HttpDownstreamConnection::read_clear;
1585 on_write_ = &HttpDownstreamConnection::write_first;
1586
1587 return 0;
1588 }
1589
on_read()1590 int HttpDownstreamConnection::on_read() { return on_read_(*this); }
1591
on_write()1592 int HttpDownstreamConnection::on_write() { return on_write_(*this); }
1593
on_upstream_change(Upstream * upstream)1594 void HttpDownstreamConnection::on_upstream_change(Upstream *upstream) {}
1595
signal_write()1596 void HttpDownstreamConnection::signal_write() { signal_write_(*this); }
1597
actual_signal_write()1598 int HttpDownstreamConnection::actual_signal_write() {
1599 ev_feed_event(conn_.loop, &conn_.wev, EV_WRITE);
1600 return 0;
1601 }
1602
noop()1603 int HttpDownstreamConnection::noop() { return 0; }
1604
1605 const std::shared_ptr<DownstreamAddrGroup> &
get_downstream_addr_group() const1606 HttpDownstreamConnection::get_downstream_addr_group() const {
1607 return group_;
1608 }
1609
get_addr() const1610 DownstreamAddr *HttpDownstreamConnection::get_addr() const { return addr_; }
1611
poolable() const1612 bool HttpDownstreamConnection::poolable() const {
1613 return !group_->retired && reusable_;
1614 }
1615
get_raddr() const1616 const Address *HttpDownstreamConnection::get_raddr() const { return raddr_; }
1617
1618 } // namespace shrpx
1619