walkero
/
nghttp2
1
0
Fork 0
Code Issues Pull Requests Projects Releases 1 Wiki Activity
nghttp2/src/shrpx_http_downstream_conne...

1498 lines
40 KiB
C++
Raw Blame History

/*
* nghttp2 - HTTP/2 C Library
*
* Copyright (c) 2012 Tatsuhiro Tsujikawa
*
* Permission is hereby granted, free of charge, to any person obtaining
* a copy of this software and associated documentation files (the
* "Software"), to deal in the Software without restriction, including
* without limitation the rights to use, copy, modify, merge, publish,
* distribute, sublicense, and/or sell copies of the Software, and to
* permit persons to whom the Software is furnished to do so, subject to
* the following conditions:
*
* The above copyright notice and this permission notice shall be
* included in all copies or substantial portions of the Software.
*
* THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
* EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
* MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
* NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
* LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
* OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
* WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
*/
#include "shrpx_http_downstream_connection.h"
#include "shrpx_client_handler.h"
#include "shrpx_upstream.h"
#include "shrpx_downstream.h"
#include "shrpx_config.h"
#include "shrpx_error.h"
#include "shrpx_http.h"
#include "shrpx_log_config.h"
#include "shrpx_connect_blocker.h"
#include "shrpx_downstream_connection_pool.h"
#include "shrpx_worker.h"
#include "shrpx_http2_session.h"
#include "shrpx_tls.h"
#include "shrpx_log.h"
#include "http2.h"
#include "util.h"
using namespace nghttp2;
namespace shrpx {
namespace {
void timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
auto conn = static_cast<Connection *>(w->data);
auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
if (w == &conn->rt && !conn->expired_rt()) {
return;
}
if (LOG_ENABLED(INFO)) {
DCLOG(INFO, dconn) << "Time out";
}
auto downstream = dconn->get_downstream();
auto upstream = downstream->get_upstream();
auto handler = upstream->get_client_handler();
auto &resp = downstream->response();
// Do this so that dconn is not pooled
resp.connection_close = true;
if (upstream->downstream_error(dconn, Downstream::EVENT_TIMEOUT) != 0) {
delete handler;
}
}
} // namespace
namespace {
void retry_downstream_connection(Downstream *downstream,
unsigned int status_code) {
auto upstream = downstream->get_upstream();
auto handler = upstream->get_client_handler();
downstream->add_retry();
if (downstream->no_more_retry()) {
delete handler;
return;
}
downstream->pop_downstream_connection();
int rv;
auto ndconn = handler->get_downstream_connection(rv, downstream);
if (ndconn) {
if (downstream->attach_downstream_connection(std::move(ndconn)) == 0 &&
downstream->push_request_headers() == 0) {
return;
}
}
downstream->set_request_state(Downstream::CONNECT_FAIL);
if (rv == SHRPX_ERR_TLS_REQUIRED) {
rv = upstream->on_downstream_abort_request_with_https_redirect(downstream);
} else {
rv = upstream->on_downstream_abort_request(downstream, status_code);
}
if (rv != 0) {
delete handler;
}
}
} // namespace
namespace {
void connect_timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
auto conn = static_cast<Connection *>(w->data);
auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
auto addr = dconn->get_addr();
auto raddr = dconn->get_raddr();
DCLOG(WARN, dconn) << "Connect time out; addr="
<< util::to_numeric_addr(raddr);
downstream_failure(addr, raddr);
auto downstream = dconn->get_downstream();
retry_downstream_connection(downstream, 504);
}
} // namespace
namespace {
void readcb(struct ev_loop *loop, ev_io *w, int revents) {
auto conn = static_cast<Connection *>(w->data);
auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
auto downstream = dconn->get_downstream();
auto upstream = downstream->get_upstream();
auto handler = upstream->get_client_handler();
if (upstream->downstream_read(dconn) != 0) {
delete handler;
}
}
} // namespace
namespace {
void backend_retry(Downstream *downstream) {
retry_downstream_connection(downstream, 502);
}
} // namespace
namespace {
void writecb(struct ev_loop *loop, ev_io *w, int revents) {
int rv;
auto conn = static_cast<Connection *>(w->data);
auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
auto downstream = dconn->get_downstream();
auto upstream = downstream->get_upstream();
auto handler = upstream->get_client_handler();
rv = upstream->downstream_write(dconn);
if (rv == SHRPX_ERR_RETRY) {
backend_retry(downstream);
return;
}
if (rv != 0) {
delete handler;
}
}
} // namespace
namespace {
void connectcb(struct ev_loop *loop, ev_io *w, int revents) {
auto conn = static_cast<Connection *>(w->data);
auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
auto downstream = dconn->get_downstream();
if (dconn->connected() != 0) {
backend_retry(downstream);
return;
}
writecb(loop, w, revents);
}
} // namespace
HttpDownstreamConnection::HttpDownstreamConnection(
const std::shared_ptr<DownstreamAddrGroup> &group, size_t initial_addr_idx,
struct ev_loop *loop, Worker *worker)
: conn_(loop, -1, nullptr, worker->get_mcpool(),
worker->get_downstream_config()->timeout.write,
worker->get_downstream_config()->timeout.read, {}, {}, connectcb,
readcb, connect_timeoutcb, this,
get_config()->tls.dyn_rec.warmup_threshold,
get_config()->tls.dyn_rec.idle_timeout, PROTO_HTTP1),
on_read_(&HttpDownstreamConnection::noop),
on_write_(&HttpDownstreamConnection::noop),
signal_write_(&HttpDownstreamConnection::noop),
worker_(worker),
ssl_ctx_(worker->get_cl_ssl_ctx()),
group_(group),
addr_(nullptr),
raddr_(nullptr),
ioctrl_(&conn_.rlimit),
response_htp_{0},
initial_addr_idx_(initial_addr_idx),
reuse_first_write_done_(true),
reusable_(true) {}
HttpDownstreamConnection::~HttpDownstreamConnection() {
if (LOG_ENABLED(INFO)) {
DCLOG(INFO, this) << "Deleted";
}
if (dns_query_) {
auto dns_tracker = worker_->get_dns_tracker();
dns_tracker->cancel(dns_query_.get());
}
}
int HttpDownstreamConnection::attach_downstream(Downstream *downstream) {
int rv;
if (LOG_ENABLED(INFO)) {
DCLOG(INFO, this) << "Attaching to DOWNSTREAM:" << downstream;
}
downstream_ = downstream;
rv = initiate_connection();
if (rv != 0) {
downstream_ = nullptr;
return rv;
}
return 0;
}
int HttpDownstreamConnection::initiate_connection() {
int rv;
auto worker_blocker = worker_->get_connect_blocker();
if (worker_blocker->blocked()) {
if (LOG_ENABLED(INFO)) {
DCLOG(INFO, this)
<< "Worker wide backend connection was blocked temporarily";
}
return SHRPX_ERR_NETWORK;
}
auto &downstreamconf = *worker_->get_downstream_config();
if (conn_.fd == -1) {
auto &shared_addr = group_->shared_addr;
auto &addrs = shared_addr->addrs;
// If session affinity is enabled, we always start with address at
// initial_addr_idx_.
size_t temp_idx = initial_addr_idx_;
auto &next_downstream = shared_addr->affinity.type == AFFINITY_NONE
? shared_addr->next
: temp_idx;
auto end = next_downstream;
for (;;) {
auto check_dns_result = dns_query_.get() != nullptr;
DownstreamAddr *addr;
if (check_dns_result) {
addr = addr_;
addr_ = nullptr;
assert(addr);
assert(addr->dns);
} else {
assert(addr_ == nullptr);
if (shared_addr->affinity.type == AFFINITY_NONE) {
addr = &addrs[next_downstream];
if (++next_downstream >= addrs.size()) {
next_downstream = 0;
}
} else {
addr = &addrs[shared_addr->affinity_hash[next_downstream].idx];
if (++next_downstream >= shared_addr->affinity_hash.size()) {
next_downstream = 0;
}
}
if (addr->proto != PROTO_HTTP1) {
if (end == next_downstream) {
return SHRPX_ERR_NETWORK;
}
continue;
}
}
auto &connect_blocker = addr->connect_blocker;
if (connect_blocker->blocked()) {
if (LOG_ENABLED(INFO)) {
DCLOG(INFO, this) << "Backend server " << addr->host << ":"
<< addr->port << " was not available temporarily";
}
if (check_dns_result) {
dns_query_.reset();
} else if (end == next_downstream) {
return SHRPX_ERR_NETWORK;
}
continue;
}
Address *raddr;
if (addr->dns) {
if (!check_dns_result) {
auto dns_query = make_unique<DNSQuery>(
addr->host, [this](int status, const Address *result) {
int rv;
if (status == DNS_STATUS_OK) {
*this->resolved_addr_ = *result;
}
rv = this->initiate_connection();
if (rv != 0) {
// This callback destroys |this|.
auto downstream = this->downstream_;
backend_retry(downstream);
}
});
auto dns_tracker = worker_->get_dns_tracker();
if (!resolved_addr_) {
resolved_addr_ = make_unique<Address>();
}
rv = dns_tracker->resolve(resolved_addr_.get(), dns_query.get());
switch (rv) {
case DNS_STATUS_ERROR:
downstream_failure(addr, nullptr);
if (end == next_downstream) {
return SHRPX_ERR_NETWORK;
}
continue;
case DNS_STATUS_RUNNING:
dns_query_ = std::move(dns_query);
// Remember current addr
addr_ = addr;
return 0;
case DNS_STATUS_OK:
break;
default:
assert(0);
}
} else {
switch (dns_query_->status) {
case DNS_STATUS_ERROR:
dns_query_.reset();
downstream_failure(addr, nullptr);
continue;
case DNS_STATUS_OK:
dns_query_.reset();
break;
default:
assert(0);
}
}
raddr = resolved_addr_.get();
util::set_port(*resolved_addr_, addr->port);
} else {
raddr = &addr->addr;
}
conn_.fd = util::create_nonblock_socket(raddr->su.storage.ss_family);
if (conn_.fd == -1) {
auto error = errno;
DCLOG(WARN, this) << "socket() failed; addr="
<< util::to_numeric_addr(raddr)
<< ", errno=" << error;
worker_blocker->on_failure();
return SHRPX_ERR_NETWORK;
}
worker_blocker->on_success();
rv = connect(conn_.fd, &raddr->su.sa, raddr->len);
if (rv != 0 && errno != EINPROGRESS) {
auto error = errno;
DCLOG(WARN, this) << "connect() failed; addr="
<< util::to_numeric_addr(raddr)
<< ", errno=" << error;
downstream_failure(addr, raddr);
close(conn_.fd);
conn_.fd = -1;
if (!check_dns_result && end == next_downstream) {
return SHRPX_ERR_NETWORK;
}
// Try again with the next downstream server
continue;
}
if (LOG_ENABLED(INFO)) {
DCLOG(INFO, this) << "Connecting to downstream server";
}
addr_ = addr;
raddr_ = raddr;
if (addr_->tls) {
assert(ssl_ctx_);
auto ssl = tls::create_ssl(ssl_ctx_);
if (!ssl) {
return -1;
}
tls::setup_downstream_http1_alpn(ssl);
conn_.set_ssl(ssl);
conn_.tls.client_session_cache = &addr_->tls_session_cache;
auto sni_name =
addr_->sni.empty() ? StringRef{addr_->host} : StringRef{addr_->sni};
if (!util::numeric_host(sni_name.c_str())) {
SSL_set_tlsext_host_name(conn_.tls.ssl, sni_name.c_str());
}
auto session = tls::reuse_tls_session(addr_->tls_session_cache);
if (session) {
SSL_set_session(conn_.tls.ssl, session);
SSL_SESSION_free(session);
}
conn_.prepare_client_handshake();
}
ev_io_set(&conn_.wev, conn_.fd, EV_WRITE);
ev_io_set(&conn_.rev, conn_.fd, EV_READ);
conn_.wlimit.startw();
break;
}
conn_.wt.repeat = downstreamconf.timeout.connect;
ev_timer_again(conn_.loop, &conn_.wt);
} else {
// we may set read timer cb to idle_timeoutcb. Reset again.
ev_set_cb(&conn_.rt, timeoutcb);
if (conn_.read_timeout < downstreamconf.timeout.read) {
conn_.read_timeout = downstreamconf.timeout.read;
conn_.last_read = ev_now(conn_.loop);
} else {
conn_.again_rt(downstreamconf.timeout.read);
}
ev_set_cb(&conn_.rev, readcb);
on_write_ = &HttpDownstreamConnection::write_reuse_first;
reuse_first_write_done_ = false;
}
http_parser_init(&response_htp_, HTTP_RESPONSE);
response_htp_.data = downstream_;
return 0;
}
int HttpDownstreamConnection::push_request_headers() {
if (downstream_->get_request_header_sent()) {
signal_write();
return 0;
}
const auto &downstream_hostport = addr_->hostport;
const auto &req = downstream_->request();
auto &balloc = downstream_->get_block_allocator();
auto connect_method = req.regular_connect_method();
auto config = get_config();
auto &httpconf = config->http;
// Set request_sent to true because we write request into buffer
// here.
downstream_->set_request_header_sent(true);
// For HTTP/1.0 request, there is no authority in request. In that
// case, we use backend server's host nonetheless.
auto authority = StringRef(downstream_hostport);
auto no_host_rewrite =
httpconf.no_host_rewrite || config->http2_proxy || connect_method;
if (no_host_rewrite && !req.authority.empty()) {
authority = req.authority;
}
downstream_->set_request_downstream_host(authority);
auto buf = downstream_->get_request_buf();
// Assume that method and request path do not contain \r\n.
auto meth = http2::to_method_string(
req.connect_proto == CONNECT_PROTO_WEBSOCKET ? HTTP_GET : req.method);
buf->append(meth);
buf->append(' ');
if (connect_method) {
buf->append(authority);
} else if (config->http2_proxy) {
// Construct absolute-form request target because we are going to
// send a request to a HTTP/1 proxy.
assert(!req.scheme.empty());
buf->append(req.scheme);
buf->append("://");
buf->append(authority);
buf->append(req.path);
} else if (req.method == HTTP_OPTIONS && req.path.empty()) {
// Server-wide OPTIONS
buf->append("*");
} else {
buf->append(req.path);
}
buf->append(" HTTP/1.1\r\nHost: ");
buf->append(authority);
buf->append("\r\n");
auto &fwdconf = httpconf.forwarded;
auto &xffconf = httpconf.xff;
auto &xfpconf = httpconf.xfp;
uint32_t build_flags =
(fwdconf.strip_incoming ? http2::HDOP_STRIP_FORWARDED : 0) |
(xffconf.strip_incoming ? http2::HDOP_STRIP_X_FORWARDED_FOR : 0) |
(xfpconf.strip_incoming ? http2::HDOP_STRIP_X_FORWARDED_PROTO : 0);
http2::build_http1_headers_from_headers(buf, req.fs.headers(), build_flags);
auto cookie = downstream_->assemble_request_cookie();
if (!cookie.empty()) {
buf->append("Cookie: ");
buf->append(cookie);
buf->append("\r\n");
}
// set transfer-encoding only when content-length is unknown and
// request body is expected.
if (req.method != HTTP_CONNECT && req.http2_expect_body &&
req.fs.content_length == -1) {
downstream_->set_chunked_request(true);
buf->append("Transfer-Encoding: chunked\r\n");
}
if (req.connection_close) {
buf->append("Connection: close\r\n");
}
if (req.connect_proto == CONNECT_PROTO_WEBSOCKET) {
// TODO Generate Sec-WebSocket-Key
buf->append("Upgrade: websocket\r\nConnection: "
"Upgrade\r\nSec-WebSocket-Key: dGhlIHNhbXBsZSBub25jZQ==\r\n");
} else if (!connect_method && req.upgrade_request) {
auto connection = req.fs.header(http2::HD_CONNECTION);
if (connection) {
buf->append("Connection: ");
buf->append((*connection).value);
buf->append("\r\n");
}
auto upgrade = req.fs.header(http2::HD_UPGRADE);
if (upgrade) {
buf->append("Upgrade: ");
buf->append((*upgrade).value);
buf->append("\r\n");
}
}
auto upstream = downstream_->get_upstream();
auto handler = upstream->get_client_handler();
auto fwd =
fwdconf.strip_incoming ? nullptr : req.fs.header(http2::HD_FORWARDED);
if (fwdconf.params) {
auto params = fwdconf.params;
if (config->http2_proxy || connect_method) {
params &= ~FORWARDED_PROTO;
}
auto value = http::create_forwarded(
balloc, params, handler->get_forwarded_by(),
handler->get_forwarded_for(), req.authority, req.scheme);
if (fwd || !value.empty()) {
buf->append("Forwarded: ");
if (fwd) {
buf->append(fwd->value);
if (!value.empty()) {
buf->append(", ");
}
}
buf->append(value);
buf->append("\r\n");
}
} else if (fwd) {
buf->append("Forwarded: ");
buf->append(fwd->value);
buf->append("\r\n");
}
auto xff = xffconf.strip_incoming ? nullptr
: req.fs.header(http2::HD_X_FORWARDED_FOR);
if (xffconf.add) {
buf->append("X-Forwarded-For: ");
if (xff) {
buf->append((*xff).value);
buf->append(", ");
}
buf->append(client_handler_->get_ipaddr());
buf->append("\r\n");
} else if (xff) {
buf->append("X-Forwarded-For: ");
buf->append((*xff).value);
buf->append("\r\n");
}
if (!config->http2_proxy && !connect_method) {
auto xfp = xfpconf.strip_incoming
? nullptr
: req.fs.header(http2::HD_X_FORWARDED_PROTO);
if (xfpconf.add) {
buf->append("X-Forwarded-Proto: ");
if (xfp) {
buf->append((*xfp).value);
buf->append(", ");
}
assert(!req.scheme.empty());
buf->append(req.scheme);
buf->append("\r\n");
} else if (xfp) {
buf->append("X-Forwarded-Proto: ");
buf->append((*xfp).value);
buf->append("\r\n");
}
}
auto via = req.fs.header(http2::HD_VIA);
if (httpconf.no_via) {
if (via) {
buf->append("Via: ");
buf->append((*via).value);
buf->append("\r\n");
}
} else {
buf->append("Via: ");
if (via) {
buf->append((*via).value);
buf->append(", ");
}
std::array<char, 16> viabuf;
auto end = http::create_via_header_value(viabuf.data(), req.http_major,
req.http_minor);
buf->append(viabuf.data(), end - viabuf.data());
buf->append("\r\n");
}
for (auto &p : httpconf.add_request_headers) {
buf->append(p.name);
buf->append(": ");
buf->append(p.value);
buf->append("\r\n");
}
buf->append("\r\n");
if (LOG_ENABLED(INFO)) {
std::string nhdrs;
for (auto chunk = buf->head; chunk; chunk = chunk->next) {
nhdrs.append(chunk->pos, chunk->last);
}
if (log_config()->errorlog_tty) {
nhdrs = http::colorizeHeaders(nhdrs.c_str());
}
DCLOG(INFO, this) << "HTTP request headers. stream_id="
<< downstream_->get_stream_id() << "\n"
<< nhdrs;
}
// Don't call signal_write() if we anticipate request body. We call
// signal_write() when we received request body chunk, and it
// enables us to send headers and data in one writev system call.
if (connect_method || req.connect_proto ||
(!req.http2_expect_body && req.fs.content_length == 0)) {
signal_write();
}
return 0;
}
int HttpDownstreamConnection::push_upload_data_chunk(const uint8_t *data,
size_t datalen) {
auto chunked = downstream_->get_chunked_request();
auto output = downstream_->get_request_buf();
if (chunked) {
auto chunk_size_hex = util::utox(datalen);
output->append(chunk_size_hex);
output->append("\r\n");
}
output->append(data, datalen);
if (chunked) {
output->append("\r\n");
}
signal_write();
return 0;
}
int HttpDownstreamConnection::end_upload_data() {
signal_write();
if (!downstream_->get_chunked_request()) {
return 0;
}
const auto &req = downstream_->request();
auto output = downstream_->get_request_buf();
const auto &trailers = req.fs.trailers();
if (trailers.empty()) {
output->append("0\r\n\r\n");
} else {
output->append("0\r\n");
http2::build_http1_headers_from_headers(output, trailers,
http2::HDOP_STRIP_ALL);
output->append("\r\n");
}
return 0;
}
namespace {
void remove_from_pool(HttpDownstreamConnection *dconn) {
auto &group = dconn->get_downstream_addr_group();
auto &shared_addr = group->shared_addr;
if (shared_addr->affinity.type == AFFINITY_NONE) {
auto &dconn_pool =
dconn->get_downstream_addr_group()->shared_addr->dconn_pool;
dconn_pool.remove_downstream_connection(dconn);
return;
}
auto addr = dconn->get_addr();
auto &dconn_pool = addr->dconn_pool;
dconn_pool->remove_downstream_connection(dconn);
}
} // namespace
namespace {
void idle_readcb(struct ev_loop *loop, ev_io *w, int revents) {
auto conn = static_cast<Connection *>(w->data);
auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
if (LOG_ENABLED(INFO)) {
DCLOG(INFO, dconn) << "Idle connection EOF";
}
remove_from_pool(dconn);
// dconn was deleted
}
} // namespace
namespace {
void idle_timeoutcb(struct ev_loop *loop, ev_timer *w, int revents) {
auto conn = static_cast<Connection *>(w->data);
auto dconn = static_cast<HttpDownstreamConnection *>(conn->data);
if (w == &conn->rt && !conn->expired_rt()) {
return;
}
if (LOG_ENABLED(INFO)) {
DCLOG(INFO, dconn) << "Idle connection timeout";
}
remove_from_pool(dconn);
// dconn was deleted
}
} // namespace
void HttpDownstreamConnection::detach_downstream(Downstream *downstream) {
if (LOG_ENABLED(INFO)) {
DCLOG(INFO, this) << "Detaching from DOWNSTREAM:" << downstream;
}
downstream_ = nullptr;
ev_set_cb(&conn_.rev, idle_readcb);
ioctrl_.force_resume_read();
auto &downstreamconf = *worker_->get_downstream_config();
ev_set_cb(&conn_.rt, idle_timeoutcb);
if (conn_.read_timeout < downstreamconf.timeout.idle_read) {
conn_.read_timeout = downstreamconf.timeout.idle_read;
conn_.last_read = ev_now(conn_.loop);
} else {
conn_.again_rt(downstreamconf.timeout.idle_read);
}
conn_.wlimit.stopw();
ev_timer_stop(conn_.loop, &conn_.wt);
}
void HttpDownstreamConnection::pause_read(IOCtrlReason reason) {
ioctrl_.pause_read(reason);
}
int HttpDownstreamConnection::resume_read(IOCtrlReason reason,
size_t consumed) {
auto &downstreamconf = *worker_->get_downstream_config();
if (downstream_->get_response_buf()->rleft() <=
downstreamconf.request_buffer_size / 2) {
ioctrl_.resume_read(reason);
}
return 0;
}
void HttpDownstreamConnection::force_resume_read() {
ioctrl_.force_resume_read();
}
namespace {
int htp_msg_begincb(http_parser *htp) {
auto downstream = static_cast<Downstream *>(htp->data);
if (downstream->get_response_state() != Downstream::INITIAL) {
return -1;
}
return 0;
}
} // namespace
namespace {
int htp_hdrs_completecb(http_parser *htp) {
auto downstream = static_cast<Downstream *>(htp->data);
auto upstream = downstream->get_upstream();
auto handler = upstream->get_client_handler();
const auto &req = downstream->request();
auto &resp = downstream->response();
int rv;
auto config = get_config();
auto &loggingconf = config->logging;
resp.http_status = htp->status_code;
resp.http_major = htp->http_major;
resp.http_minor = htp->http_minor;
if (resp.http_major > 1 || req.http_minor > 1) {
resp.http_major = 1;
resp.http_minor = 1;
return -1;
}
auto dconn = downstream->get_downstream_connection();
downstream->set_downstream_addr_group(dconn->get_downstream_addr_group());
downstream->set_addr(dconn->get_addr());
// Server MUST NOT send Transfer-Encoding with a status code 1xx or
// 204. Also server MUST NOT send Transfer-Encoding with a status
// code 200 to a CONNECT request. Same holds true with
// Content-Length.
if (resp.http_status == 204) {
if (resp.fs.header(http2::HD_TRANSFER_ENCODING)) {
return -1;
}
// Some server send content-length: 0 for 204. Until they get
// fixed, we accept, but ignore it.
// Calling parse_content_length() detects duplicated
// content-length header fields.
if (resp.fs.parse_content_length() != 0) {
return -1;
}
if (resp.fs.content_length == 0) {
auto cl = resp.fs.header(http2::HD_CONTENT_LENGTH);
assert(cl);
http2::erase_header(cl);
} else if (resp.fs.content_length != -1) {
return -1;
}
} else if (resp.http_status / 100 == 1 ||
(resp.http_status == 200 && req.regular_connect_method())) {
if (resp.fs.header(http2::HD_CONTENT_LENGTH) ||
resp.fs.header(http2::HD_TRANSFER_ENCODING)) {
return -1;
}
} else if (resp.fs.parse_content_length() != 0) {
downstream->set_response_state(Downstream::MSG_BAD_HEADER);
return -1;
}
// Check upgrade before processing non-final response, since if
// upgrade succeeded, 101 response is treated as final in nghttpx.
downstream->check_upgrade_fulfilled();
if (downstream->get_non_final_response()) {
// Reset content-length because we reuse same Downstream for the
// next response.
resp.fs.content_length = -1;
// For non-final response code, we just call
// on_downstream_header_complete() without changing response
// state.
rv = upstream->on_downstream_header_complete(downstream);
if (rv != 0) {
return -1;
}
// Ignore response body for non-final response.
return 1;
}
resp.connection_close = !http_should_keep_alive(htp);
downstream->set_response_state(Downstream::HEADER_COMPLETE);
downstream->inspect_http1_response();
if (downstream->get_upgraded()) {
// content-length must be ignored for upgraded connection.
resp.fs.content_length = -1;
resp.connection_close = true;
// transfer-encoding not applied to upgraded connection
downstream->set_chunked_response(false);
} else if (!downstream->expect_response_body()) {
downstream->set_chunked_response(false);
}
if (loggingconf.access.write_early && downstream->accesslog_ready()) {
handler->write_accesslog(downstream);
downstream->set_accesslog_written(true);
}
if (upstream->on_downstream_header_complete(downstream) != 0) {
return -1;
}
if (downstream->get_upgraded()) {
// Upgrade complete, read until EOF in both ends
if (upstream->resume_read(SHRPX_NO_BUFFER, downstream, 0) != 0) {
return -1;
}
downstream->set_request_state(Downstream::HEADER_COMPLETE);
if (LOG_ENABLED(INFO)) {
LOG(INFO) << "HTTP upgrade success. stream_id="
<< downstream->get_stream_id();
}
}
// Ignore the response body. HEAD response may contain
// Content-Length or Transfer-Encoding: chunked. Some server send
// 304 status code with nonzero Content-Length, but without response
// body. See
// https://tools.ietf.org/html/rfc7230#section-3.3
// TODO It seems that the cases other than HEAD are handled by
// http-parser. Need test.
return !http2::expect_response_body(req.method, resp.http_status);
}
} // namespace
namespace {
int ensure_header_field_buffer(const Downstream *downstream,
const HttpConfig &httpconf, size_t len) {
auto &resp = downstream->response();
if (resp.fs.buffer_size() + len > httpconf.response_header_field_buffer) {
if (LOG_ENABLED(INFO)) {
DLOG(INFO, downstream) << "Too large header header field size="
<< resp.fs.buffer_size() + len;
}
return -1;
}
return 0;
}
} // namespace
namespace {
int ensure_max_header_fields(const Downstream *downstream,
const HttpConfig &httpconf) {
auto &resp = downstream->response();
if (resp.fs.num_fields() >= httpconf.max_response_header_fields) {
if (LOG_ENABLED(INFO)) {
DLOG(INFO, downstream)
<< "Too many header field num=" << resp.fs.num_fields() + 1;
}
return -1;
}
return 0;
}
} // namespace
namespace {
int htp_hdr_keycb(http_parser *htp, const char *data, size_t len) {
auto downstream = static_cast<Downstream *>(htp->data);
auto &resp = downstream->response();
auto &httpconf = get_config()->http;
if (ensure_header_field_buffer(downstream, httpconf, len) != 0) {
return -1;
}
if (downstream->get_response_state() == Downstream::INITIAL) {
if (resp.fs.header_key_prev()) {
resp.fs.append_last_header_key(data, len);
} else {
if (ensure_max_header_fields(downstream, httpconf) != 0) {
return -1;
}
resp.fs.alloc_add_header_name(StringRef{data, len});
}
} else {
// trailer part
if (resp.fs.trailer_key_prev()) {
resp.fs.append_last_trailer_key(data, len);
} else {
if (ensure_max_header_fields(downstream, httpconf) != 0) {
// Could not ignore this trailer field easily, since we may
// get its value in htp_hdr_valcb, and it will be added to
// wrong place or crash if trailer fields are currently empty.
return -1;
}
resp.fs.alloc_add_trailer_name(StringRef{data, len});
}
}
return 0;
}
} // namespace
namespace {
int htp_hdr_valcb(http_parser *htp, const char *data, size_t len) {
auto downstream = static_cast<Downstream *>(htp->data);
auto &resp = downstream->response();
auto &httpconf = get_config()->http;
if (ensure_header_field_buffer(downstream, httpconf, len) != 0) {
return -1;
}
if (downstream->get_response_state() == Downstream::INITIAL) {
resp.fs.append_last_header_value(data, len);
} else {
resp.fs.append_last_trailer_value(data, len);
}
return 0;
}
} // namespace
namespace {
int htp_bodycb(http_parser *htp, const char *data, size_t len) {
auto downstream = static_cast<Downstream *>(htp->data);
auto &resp = downstream->response();
resp.recv_body_length += len;
return downstream->get_upstream()->on_downstream_body(
downstream, reinterpret_cast<const uint8_t *>(data), len, true);
}
} // namespace
namespace {
int htp_msg_completecb(http_parser *htp) {
auto downstream = static_cast<Downstream *>(htp->data);
// http-parser does not treat "200 connection established" response
// against CONNECT request, and in that case, this function is not
// called. But if HTTP Upgrade is made (e.g., WebSocket), this
// function is called, and http_parser_execute() returns just after
// that.
if (downstream->get_upgraded()) {
return 0;
}
if (downstream->get_non_final_response()) {
downstream->reset_response();
return 0;
}
downstream->set_response_state(Downstream::MSG_COMPLETE);
// Block reading another response message from (broken?)
// server. This callback is not called if the connection is
// tunneled.
downstream->pause_read(SHRPX_MSG_BLOCK);
return downstream->get_upstream()->on_downstream_body_complete(downstream);
}
} // namespace
namespace {
constexpr http_parser_settings htp_hooks = {
htp_msg_begincb, // http_cb on_message_begin;
nullptr, // http_data_cb on_url;
nullptr, // http_data_cb on_status;
htp_hdr_keycb, // http_data_cb on_header_field;
htp_hdr_valcb, // http_data_cb on_header_value;
htp_hdrs_completecb, // http_cb on_headers_complete;
htp_bodycb, // http_data_cb on_body;
htp_msg_completecb // http_cb on_message_complete;
};
} // namespace
int HttpDownstreamConnection::write_reuse_first() {
int rv;
if (conn_.tls.ssl) {
rv = write_tls();
} else {
rv = write_clear();
}
if (rv != 0) {
return SHRPX_ERR_RETRY;
}
if (conn_.tls.ssl) {
on_write_ = &HttpDownstreamConnection::write_tls;
} else {
on_write_ = &HttpDownstreamConnection::write_clear;
}
reuse_first_write_done_ = true;
return 0;
}
int HttpDownstreamConnection::read_clear() {
conn_.last_read = ev_now(conn_.loop);
std::array<uint8_t, 16_k> buf;
int rv;
for (;;) {
auto nread = conn_.read_clear(buf.data(), buf.size());
if (nread == 0) {
return 0;
}
if (nread < 0) {
return nread;
}
rv = process_input(buf.data(), nread);
if (rv != 0) {
return rv;
}
if (!ev_is_active(&conn_.rev)) {
return 0;
}
}
}
int HttpDownstreamConnection::write_clear() {
conn_.last_read = ev_now(conn_.loop);
auto upstream = downstream_->get_upstream();
auto input = downstream_->get_request_buf();
std::array<struct iovec, MAX_WR_IOVCNT> iov;
while (input->rleft() > 0) {
auto iovcnt = input->riovec(iov.data(), iov.size());
auto nwrite = conn_.writev_clear(iov.data(), iovcnt);
if (nwrite == 0) {
return 0;
}
if (nwrite < 0) {
if (!reuse_first_write_done_) {
return nwrite;
}
// We may have pending data in receive buffer which may contain
// part of response body. So keep reading. Invoke read event
// to get read(2) error just in case.
ev_feed_event(conn_.loop, &conn_.rev, EV_READ);
on_write_ = &HttpDownstreamConnection::noop;
reusable_ = false;
break;
}
input->drain(nwrite);
}
conn_.wlimit.stopw();
ev_timer_stop(conn_.loop, &conn_.wt);
if (input->rleft() == 0) {
auto &req = downstream_->request();
upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
req.unconsumed_body_length);
}
return 0;
}
int HttpDownstreamConnection::tls_handshake() {
ERR_clear_error();
conn_.last_read = ev_now(conn_.loop);
auto rv = conn_.tls_handshake();
if (rv == SHRPX_ERR_INPROGRESS) {
return 0;
}
if (rv < 0) {
downstream_failure(addr_, raddr_);
return rv;
}
if (LOG_ENABLED(INFO)) {
DCLOG(INFO, this) << "SSL/TLS handshake completed";
}
if (!get_config()->tls.insecure &&
tls::check_cert(conn_.tls.ssl, addr_, raddr_) != 0) {
downstream_failure(addr_, raddr_);
return -1;
}
auto &connect_blocker = addr_->connect_blocker;
signal_write_ = &HttpDownstreamConnection::actual_signal_write;
connect_blocker->on_success();
ev_set_cb(&conn_.rt, timeoutcb);
ev_set_cb(&conn_.wt, timeoutcb);
on_read_ = &HttpDownstreamConnection::read_tls;
on_write_ = &HttpDownstreamConnection::write_tls;
// TODO Check negotiated ALPN
return on_write();
}
int HttpDownstreamConnection::read_tls() {
conn_.last_read = ev_now(conn_.loop);
ERR_clear_error();
std::array<uint8_t, 16_k> buf;
int rv;
for (;;) {
auto nread = conn_.read_tls(buf.data(), buf.size());
if (nread == 0) {
return 0;
}
if (nread < 0) {
return nread;
}
rv = process_input(buf.data(), nread);
if (rv != 0) {
return rv;
}
if (!ev_is_active(&conn_.rev)) {
return 0;
}
}
}
int HttpDownstreamConnection::write_tls() {
conn_.last_read = ev_now(conn_.loop);
ERR_clear_error();
auto upstream = downstream_->get_upstream();
auto input = downstream_->get_request_buf();
struct iovec iov;
while (input->rleft() > 0) {
auto iovcnt = input->riovec(&iov, 1);
if (iovcnt != 1) {
assert(0);
return -1;
}
auto nwrite = conn_.write_tls(iov.iov_base, iov.iov_len);
if (nwrite == 0) {
return 0;
}
if (nwrite < 0) {
if (!reuse_first_write_done_) {
return nwrite;
}
// We may have pending data in receive buffer which may contain
// part of response body. So keep reading. Invoke read event
// to get read(2) error just in case.
ev_feed_event(conn_.loop, &conn_.rev, EV_READ);
on_write_ = &HttpDownstreamConnection::noop;
reusable_ = false;
break;
}
input->drain(nwrite);
}
conn_.wlimit.stopw();
ev_timer_stop(conn_.loop, &conn_.wt);
if (input->rleft() == 0) {
auto &req = downstream_->request();
upstream->resume_read(SHRPX_NO_BUFFER, downstream_,
req.unconsumed_body_length);
}
return 0;
}
int HttpDownstreamConnection::process_input(const uint8_t *data,
size_t datalen) {
int rv;
if (downstream_->get_upgraded()) {
// For upgraded connection, just pass data to the upstream.
rv = downstream_->get_upstream()->on_downstream_body(downstream_, data,
datalen, true);
if (rv != 0) {
return rv;
}
if (downstream_->response_buf_full()) {
downstream_->pause_read(SHRPX_NO_BUFFER);
return 0;
}
return 0;
}
auto nproc =
http_parser_execute(&response_htp_, &htp_hooks,
reinterpret_cast<const char *>(data), datalen);
auto htperr = HTTP_PARSER_ERRNO(&response_htp_);
if (htperr != HPE_OK) {
// Handling early return (in other words, response was hijacked by
// mruby scripting).
if (downstream_->get_response_state() == Downstream::MSG_COMPLETE) {
return SHRPX_ERR_DCONN_CANCELED;
}
if (LOG_ENABLED(INFO)) {
DCLOG(INFO, this) << "HTTP parser failure: "
<< "(" << http_errno_name(htperr) << ") "
<< http_errno_description(htperr);
}
return -1;
}
if (downstream_->get_upgraded()) {
if (nproc < datalen) {
// Data from data + nproc are for upgraded protocol.
rv = downstream_->get_upstream()->on_downstream_body(
downstream_, data + nproc, datalen - nproc, true);
if (rv != 0) {
return rv;
}
if (downstream_->response_buf_full()) {
downstream_->pause_read(SHRPX_NO_BUFFER);
return 0;
}
}
return 0;
}
if (downstream_->response_buf_full()) {
downstream_->pause_read(SHRPX_NO_BUFFER);
return 0;
}
return 0;
}
int HttpDownstreamConnection::connected() {
auto &connect_blocker = addr_->connect_blocker;
auto sock_error = util::get_socket_error(conn_.fd);
if (sock_error != 0) {
conn_.wlimit.stopw();
DCLOG(WARN, this) << "Backend connect failed; addr="
<< util::to_numeric_addr(raddr_)
<< ": errno=" << sock_error;
downstream_failure(addr_, raddr_);
return -1;
}
if (LOG_ENABLED(INFO)) {
DCLOG(INFO, this) << "Connected to downstream host";
}
auto &downstreamconf = *get_config()->conn.downstream;
// Reset timeout for write. Previously, we set timeout for connect.
conn_.wt.repeat = downstreamconf.timeout.write;
ev_timer_again(conn_.loop, &conn_.wt);
conn_.rlimit.startw();
conn_.again_rt();
ev_set_cb(&conn_.wev, writecb);
if (conn_.tls.ssl) {
on_read_ = &HttpDownstreamConnection::tls_handshake;
on_write_ = &HttpDownstreamConnection::tls_handshake;
return 0;
}
signal_write_ = &HttpDownstreamConnection::actual_signal_write;
connect_blocker->on_success();
ev_set_cb(&conn_.rt, timeoutcb);
ev_set_cb(&conn_.wt, timeoutcb);
on_read_ = &HttpDownstreamConnection::read_clear;
on_write_ = &HttpDownstreamConnection::write_clear;
return 0;
}
int HttpDownstreamConnection::on_read() { return on_read_(*this); }
int HttpDownstreamConnection::on_write() { return on_write_(*this); }
void HttpDownstreamConnection::on_upstream_change(Upstream *upstream) {}
void HttpDownstreamConnection::signal_write() { signal_write_(*this); }
int HttpDownstreamConnection::actual_signal_write() {
ev_feed_event(conn_.loop, &conn_.wev, EV_WRITE);
return 0;
}
int HttpDownstreamConnection::noop() { return 0; }
const std::shared_ptr<DownstreamAddrGroup> &
HttpDownstreamConnection::get_downstream_addr_group() const {
return group_;
}
DownstreamAddr *HttpDownstreamConnection::get_addr() const { return addr_; }
bool HttpDownstreamConnection::poolable() const {
return !group_->retired && reusable_;
}
const Address *HttpDownstreamConnection::get_raddr() const { return raddr_; }
} // namespace shrpx
Reference in New Issue View Git Blame Copy Permalink