X-Git-Url: https://git.sesse.net/?p=cubemap;a=blobdiff_plain;f=server.cpp;h=7f07483390a8b2846013e594668831b48d5b884b;hp=a418d1c660d409b0ea85162696f9a33de01dc330;hb=b10d63c534fda113e65e24d252e509e616067ef9;hpb=52d37b2815f4b144fbac899b7f9548353a6fbe3c diff --git a/server.cpp b/server.cpp index a418d1c..7f07483 100644 --- a/server.cpp +++ b/server.cpp @@ -17,6 +17,9 @@ #include #include +#include "tlse.h" + +#include "acceptor.h" #include "accesslog.h" #include "log.h" #include "metacube2.h" @@ -140,7 +143,7 @@ void Server::do_work() } timeout_time.tv_sec -= REQUEST_READ_TIMEOUT_SEC; while (!clients_ordered_by_connect_time.empty()) { - pair &connect_time_and_fd = clients_ordered_by_connect_time.front(); + const pair &connect_time_and_fd = clients_ordered_by_connect_time.front(); // See if we have reached the end of clients to process. if (is_earlier(timeout_time, connect_time_and_fd.first)) { @@ -149,7 +152,7 @@ void Server::do_work() // If this client doesn't exist anymore, just ignore it // (it was deleted earlier). - std::map::iterator client_it = clients.find(connect_time_and_fd.second); + map::iterator client_it = clients.find(connect_time_and_fd.second); if (client_it == clients.end()) { clients_ordered_by_connect_time.pop(); continue; @@ -204,14 +207,15 @@ CubemapStateProto Server::serialize() return serialized; } -void Server::add_client_deferred(int sock) +void Server::add_client_deferred(int sock, Acceptor *acceptor) { MutexLock lock(&queued_clients_mutex); - queued_add_clients.push_back(sock); + queued_add_clients.push_back(std::make_pair(sock, acceptor)); } -void Server::add_client(int sock) +void Server::add_client(int sock, Acceptor *acceptor) { + const bool is_tls = acceptor->is_tls(); pair::iterator, bool> ret = clients.insert(make_pair(sock, Client(sock))); assert(ret.second == true); // Should not already exist. @@ -230,13 +234,32 @@ void Server::add_client(int sock) // Start listening on data from this socket. epoll_event ev; - ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP; + if (is_tls) { + // Even in the initial state (READING_REQUEST), TLS needs to + // send data for the handshake, and thus might end up needing + // to know about EPOLLOUT. + ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLRDHUP; + } else { + // EPOLLOUT will be added once we go out of READING_REQUEST. + ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP; + } ev.data.u64 = reinterpret_cast(client_ptr); if (epoll_ctl(epoll_fd, EPOLL_CTL_ADD, sock, &ev) == -1) { log_perror("epoll_ctl(EPOLL_CTL_ADD)"); exit(1); } + if (is_tls) { + assert(tls_server_contexts.count(acceptor)); + client_ptr->tls_context = tls_accept(tls_server_contexts[acceptor]); + if (client_ptr->tls_context == NULL) { + log(ERROR, "tls_accept() failed"); + close_client(client_ptr); + return; + } + tls_make_exportable(client_ptr->tls_context, 1); + } + process_client(client_ptr); } @@ -264,7 +287,12 @@ void Server::add_client_from_serialized(const ClientProto &client) // Start listening on data from this socket. epoll_event ev; if (client.state() == Client::READING_REQUEST) { - ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP; + // See the corresponding comment in Server::add_client(). + if (client.has_tls_context()) { + ev.events = EPOLLIN | EPOLLOUT | EPOLLET | EPOLLRDHUP; + } else { + ev.events = EPOLLIN | EPOLLET | EPOLLRDHUP; + } } else { // If we don't have more data for this client, we'll be putting it into // the sleeping array again soon. @@ -286,27 +314,27 @@ void Server::add_client_from_serialized(const ClientProto &client) } } -int Server::lookup_stream_by_url(const std::string &url) const +int Server::lookup_stream_by_url(const string &url) const { - map::const_iterator url_it = url_map.find(url); - if (url_it == url_map.end()) { + map::const_iterator stream_url_it = stream_url_map.find(url); + if (stream_url_it == stream_url_map.end()) { return -1; } - return url_it->second; + return stream_url_it->second; } -int Server::add_stream(const string &url, size_t backlog_size, size_t prebuffering_bytes, Stream::Encoding encoding) +int Server::add_stream(const string &url, size_t backlog_size, size_t prebuffering_bytes, Stream::Encoding encoding, Stream::Encoding src_encoding) { MutexLock lock(&mutex); - url_map.insert(make_pair(url, streams.size())); - streams.push_back(new Stream(url, backlog_size, prebuffering_bytes, encoding)); + stream_url_map.insert(make_pair(url, streams.size())); + streams.push_back(new Stream(url, backlog_size, prebuffering_bytes, encoding, src_encoding)); return streams.size() - 1; } int Server::add_stream_from_serialized(const StreamProto &stream, int data_fd) { MutexLock lock(&mutex); - url_map.insert(make_pair(stream.url(), streams.size())); + stream_url_map.insert(make_pair(stream.url(), streams.size())); streams.push_back(new Stream(stream, data_fd)); return streams.size() - 1; } @@ -331,6 +359,13 @@ void Server::set_encoding(int stream_index, Stream::Encoding encoding) assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); streams[stream_index]->encoding = encoding; } + +void Server::set_src_encoding(int stream_index, Stream::Encoding encoding) +{ + MutexLock lock(&mutex); + assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); + streams[stream_index]->src_encoding = encoding; +} void Server::set_header(int stream_index, const string &http_header, const string &stream_header) { @@ -359,10 +394,35 @@ void Server::set_pacing_rate(int stream_index, uint32_t pacing_rate) streams[stream_index]->pacing_rate = pacing_rate; } -void Server::add_data_deferred(int stream_index, const char *data, size_t bytes, StreamStartSuitability suitable_for_stream_start) +void Server::add_gen204(const std::string &url, const std::string &allow_origin) +{ + MutexLock lock(&mutex); + assert(clients.empty()); + ping_url_map[url] = allow_origin; +} + +void Server::create_tls_context_for_acceptor(const Acceptor *acceptor) +{ + assert(acceptor->is_tls()); + + bool is_server = true; + TLSContext *server_context = tls_create_context(is_server, TLS_V12); + + const string &cert = acceptor->get_certificate_chain(); + int num_cert = tls_load_certificates(server_context, reinterpret_cast(cert.data()), cert.size()); + assert(num_cert > 0); // Should have been checked by config earlier. + + const string &key = acceptor->get_private_key(); + int num_key = tls_load_private_key(server_context, reinterpret_cast(key.data()), key.size()); + assert(num_key > 0); // Should have been checked by config earlier. + + tls_server_contexts.insert(make_pair(acceptor, server_context)); +} + +void Server::add_data_deferred(int stream_index, const char *data, size_t bytes, uint16_t metacube_flags) { assert(stream_index >= 0 && stream_index < ssize_t(streams.size())); - streams[stream_index]->add_data_deferred(data, bytes, suitable_for_stream_start); + streams[stream_index]->add_data_deferred(data, bytes, metacube_flags); } // See the .h file for postconditions after this function. @@ -370,28 +430,29 @@ void Server::process_client(Client *client) { switch (client->state) { case Client::READING_REQUEST: { + if (client->tls_context != NULL) { + if (send_pending_tls_data(client)) { + // send_pending_tls_data() hit postconditions #1 or #4. + return; + } + } + read_request_again: // Try to read more of the request. char buf[1024]; int ret; - do { - ret = read(client->sock, buf, sizeof(buf)); - } while (ret == -1 && errno == EINTR); - - if (ret == -1 && errno == EAGAIN) { - // No more data right now. Nothing to do. - // This is postcondition #2. - return; - } - if (ret == -1) { - log_perror("read"); - close_client(client); - return; - } - if (ret == 0) { - // OK, the socket is closed. - close_client(client); - return; + if (client->tls_context == NULL) { + ret = read_nontls_data(client, buf, sizeof(buf)); + if (ret == -1) { + // read_nontls_data() hit postconditions #1 or #2. + return; + } + } else { + ret = read_tls_data(client, buf, sizeof(buf)); + if (ret == -1) { + // read_tls_data() hit postconditions #1, #2 or #4. + return; + } } RequestParseStatus status = wait_for_double_newline(&client->request, buf, ret); @@ -415,25 +476,43 @@ read_request_again: assert(status == RP_FINISHED); + if (client->tls_context && !client->in_ktls_mode && tls_established(client->tls_context)) { + // We're ready to enter kTLS mode, unless we still have some + // handshake data to send (which then must be sent as non-kTLS). + if (send_pending_tls_data(client)) { + // send_pending_tls_data() hit postconditions #1 or #4. + return; + } + ret = tls_make_ktls(client->tls_context, client->sock); + if (ret < 0) { + log_tls_error("tls_make_ktls", ret); + close_client(client); + return; + } + client->in_ktls_mode = true; + } + int error_code = parse_request(client); if (error_code == 200) { construct_header(client); + } else if (error_code == 204) { + construct_204(client); } else { construct_error(client, error_code); } // We've changed states, so fall through. - assert(client->state == Client::SENDING_ERROR || + assert(client->state == Client::SENDING_SHORT_RESPONSE || client->state == Client::SENDING_HEADER); } - case Client::SENDING_ERROR: + case Client::SENDING_SHORT_RESPONSE: case Client::SENDING_HEADER: { -sending_header_or_error_again: +sending_header_or_short_response_again: int ret; do { ret = write(client->sock, - client->header_or_error.data() + client->header_or_error_bytes_sent, - client->header_or_error.size() - client->header_or_error_bytes_sent); + client->header_or_short_response.data() + client->header_or_short_response_bytes_sent, + client->header_or_short_response.size() - client->header_or_short_response_bytes_sent); } while (ret == -1 && errno == EINTR); if (ret == -1 && errno == EAGAIN) { @@ -451,39 +530,63 @@ sending_header_or_error_again: return; } - client->header_or_error_bytes_sent += ret; - assert(client->header_or_error_bytes_sent <= client->header_or_error.size()); + client->header_or_short_response_bytes_sent += ret; + assert(client->header_or_short_response_bytes_sent <= client->header_or_short_response.size()); - if (client->header_or_error_bytes_sent < client->header_or_error.size()) { + if (client->header_or_short_response_bytes_sent < client->header_or_short_response.size()) { // We haven't sent all yet. Fine; go another round. - goto sending_header_or_error_again; + goto sending_header_or_short_response_again; } // We're done sending the header or error! Clear it to release some memory. - client->header_or_error.clear(); + client->header_or_short_response.clear(); - if (client->state == Client::SENDING_ERROR) { + if (client->state == Client::SENDING_SHORT_RESPONSE) { // We're done sending the error, so now close. // This is postcondition #1. close_client(client); return; } - // Start sending from the first keyframe we get. In other - // words, we won't send any of the backlog, but we'll start - // sending immediately as we get the next keyframe block. - // This is postcondition #3. + Stream *stream = client->stream; if (client->stream_pos == size_t(-2)) { - client->stream_pos = std::min( - client->stream->bytes_received - client->stream->backlog_size, + // Start sending from the beginning of the backlog. + client->stream_pos = min( + stream->bytes_received - stream->backlog_size, 0); client->state = Client::SENDING_DATA; - } else { - client->stream_pos = client->stream->bytes_received; + goto sending_data; + } else if (stream->prebuffering_bytes == 0) { + // Start sending from the first keyframe we get. In other + // words, we won't send any of the backlog, but we'll start + // sending immediately as we get the next keyframe block. + // Note that this is functionally identical to the next if branch, + // except that we save a binary search. + client->stream_pos = stream->bytes_received; client->state = Client::WAITING_FOR_KEYFRAME; + } else { + // We're not going to send anything to the client before we have + // N bytes. However, this wait might be boring; we can just as well + // use it to send older data if we have it. We use lower_bound() + // so that we are conservative and never add extra latency over just + // waiting (assuming CBR or nearly so); otherwise, we could want e.g. + // 100 kB prebuffer but end up sending a 10 MB GOP. + deque::const_iterator starting_point_it = + lower_bound(stream->suitable_starting_points.begin(), + stream->suitable_starting_points.end(), + stream->bytes_received - stream->prebuffering_bytes); + if (starting_point_it == stream->suitable_starting_points.end()) { + // None found. Just put us at the end, and then wait for the + // first keyframe to appear. + client->stream_pos = stream->bytes_received; + client->state = Client::WAITING_FOR_KEYFRAME; + } else { + client->stream_pos = *starting_point_it; + client->state = Client::PREBUFFERING; + goto prebuffering; + } } - client->stream->put_client_to_sleep(client); - return; + // Fall through. } case Client::WAITING_FOR_KEYFRAME: { Stream *stream = client->stream; @@ -500,6 +603,7 @@ sending_header_or_error_again: // Fall through. } case Client::PREBUFFERING: { +prebuffering: Stream *stream = client->stream; size_t bytes_to_send = stream->bytes_received - client->stream_pos; assert(bytes_to_send <= stream->backlog_size); @@ -513,6 +617,7 @@ sending_header_or_error_again: // Fall through. } case Client::SENDING_DATA: { +sending_data: skip_lost_data(client); Stream *stream = client->stream; @@ -565,6 +670,133 @@ sending_data_again: } } +bool Server::send_pending_tls_data(Client *client) +{ + // See if there's data from the TLS library to write. + if (client->tls_data_to_send == NULL) { + client->tls_data_to_send = tls_get_write_buffer(client->tls_context, &client->tls_data_left_to_send); + if (client->tls_data_to_send == NULL) { + // Really no data to send. + return false; + } + } + +send_data_again: + int ret; + do { + ret = write(client->sock, client->tls_data_to_send, client->tls_data_left_to_send); + } while (ret == -1 && errno == EINTR); + assert(ret < 0 || size_t(ret) <= client->tls_data_left_to_send); + + if (ret == -1 && errno == EAGAIN) { + // We're out of socket space, so now we're at the “low edge” of epoll's + // edge triggering. epoll will tell us when there is more room, so for now, + // just return. + // This is postcondition #4. + return true; + } + if (ret == -1) { + // Error! Postcondition #1. + log_perror("write"); + close_client(client); + return true; + } + if (ret > 0 && size_t(ret) == client->tls_data_left_to_send) { + // All data has been sent, so we don't need to go to sleep. + tls_buffer_clear(client->tls_context); + client->tls_data_to_send = NULL; + return false; + } + + // More data to send, so try again. + client->tls_data_to_send += ret; + client->tls_data_left_to_send -= ret; + goto send_data_again; +} + +int Server::read_nontls_data(Client *client, char *buf, size_t max_size) +{ + int ret; + do { + ret = read(client->sock, buf, max_size); + } while (ret == -1 && errno == EINTR); + + if (ret == -1 && errno == EAGAIN) { + // No more data right now. Nothing to do. + // This is postcondition #2. + return -1; + } + if (ret == -1) { + log_perror("read"); + close_client(client); + return -1; + } + if (ret == 0) { + // OK, the socket is closed. + close_client(client); + return -1; + } + + return ret; +} + +int Server::read_tls_data(Client *client, char *buf, size_t max_size) +{ +read_again: + int ret; + do { + ret = read(client->sock, buf, max_size); + } while (ret == -1 && errno == EINTR); + + if (ret == -1 && errno == EAGAIN) { + // No more data right now. Nothing to do. + // This is postcondition #2. + return -1; + } + if (ret == -1) { + log_perror("read"); + close_client(client); + return -1; + } + if (ret == 0) { + // OK, the socket is closed. + close_client(client); + return -1; + } + + // Give it to the TLS library. + int err = tls_consume_stream(client->tls_context, reinterpret_cast(buf), ret, nullptr); + if (err < 0) { + log_tls_error("tls_consume_stream", err); + close_client(client); + return -1; + } + if (err == 0) { + // Not consumed any data. See if we can read more. + goto read_again; + } + + // Read any decrypted data available for us. (We can reuse buf, since it's free now.) + ret = tls_read(client->tls_context, reinterpret_cast(buf), max_size); + if (ret == 0) { + // No decrypted data for us yet, but there might be some more handshaking + // to send. Do that if needed, then look for more data. + if (send_pending_tls_data(client)) { + // send_pending_tls_data() hit postconditions #1 or #4. + return -1; + } + goto read_again; + } + if (ret < 0) { + log_tls_error("tls_read", ret); + close_client(client); + return -1; + } + + assert(ret > 0); + return ret; +} + // See if there's some data we've lost. Ideally, we should drop to a block boundary, // but resync will be the mux's problem. void Server::skip_lost_data(Client *client) @@ -610,25 +842,29 @@ int Server::parse_request(Client *client) } string url = request_tokens[1]; - if (url.find("?backlog") == url.size() - 8) { + client->url = url; + if (url.size() > 8 && url.find("?backlog") == url.size() - 8) { client->stream_pos = -2; url = url.substr(0, url.size() - 8); } else { client->stream_pos = -1; } - map::const_iterator url_map_it = url_map.find(url); - if (url_map_it == url_map.end()) { - return 404; // Not found. + map::const_iterator stream_url_map_it = stream_url_map.find(url); + if (stream_url_map_it == stream_url_map.end()) { + map::const_iterator ping_url_map_it = ping_url_map.find(url); + if (ping_url_map_it == ping_url_map.end()) { + return 404; // Not found. + } else { + return 204; // No error. + } } - Stream *stream = streams[url_map_it->second]; + Stream *stream = streams[stream_url_map_it->second]; if (stream->http_header.empty()) { return 503; // Service unavailable. } - client->url = request_tokens[1]; - client->stream = stream; if (setsockopt(client->sock, SOL_SOCKET, SO_MAX_PACING_RATE, &client->stream->pacing_rate, sizeof(client->stream->pacing_rate)) == -1) { if (client->stream->pacing_rate != ~0U) { @@ -644,11 +880,11 @@ void Server::construct_header(Client *client) { Stream *stream = client->stream; if (stream->encoding == Stream::STREAM_ENCODING_RAW) { - client->header_or_error = stream->http_header + + client->header_or_short_response = stream->http_header + "\r\n" + stream->stream_header; } else if (stream->encoding == Stream::STREAM_ENCODING_METACUBE) { - client->header_or_error = stream->http_header + + client->header_or_short_response = stream->http_header + "Content-encoding: metacube\r\n" + "\r\n"; if (!stream->stream_header.empty()) { @@ -657,10 +893,10 @@ void Server::construct_header(Client *client) hdr.size = htonl(stream->stream_header.size()); hdr.flags = htons(METACUBE_FLAGS_HEADER); hdr.csum = htons(metacube2_compute_crc(&hdr)); - client->header_or_error.append( + client->header_or_short_response.append( string(reinterpret_cast(&hdr), sizeof(hdr))); } - client->header_or_error.append(stream->stream_header); + client->header_or_short_response.append(stream->stream_header); } else { assert(false); } @@ -683,10 +919,42 @@ void Server::construct_error(Client *client, int error_code) char error[256]; snprintf(error, 256, "HTTP/1.0 %d Error\r\nContent-type: text/plain\r\n\r\nSomething went wrong. Sorry.\r\n", error_code); - client->header_or_error = error; + client->header_or_short_response = error; + + // Switch states. + client->state = Client::SENDING_SHORT_RESPONSE; + + epoll_event ev; + ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP; + ev.data.u64 = reinterpret_cast(client); + + if (epoll_ctl(epoll_fd, EPOLL_CTL_MOD, client->sock, &ev) == -1) { + log_perror("epoll_ctl(EPOLL_CTL_MOD)"); + exit(1); + } +} + +void Server::construct_204(Client *client) +{ + map::const_iterator ping_url_map_it = ping_url_map.find(client->url); + assert(ping_url_map_it != ping_url_map.end()); + + if (ping_url_map_it->second.empty()) { + client->header_or_short_response = + "HTTP/1.0 204 No Content\r\n" + "\r\n"; + } else { + char response[256]; + snprintf(response, 256, + "HTTP/1.0 204 No Content\r\n" + "Access-Control-Allow-Origin: %s\r\n" + "\r\n", + ping_url_map_it->second.c_str()); + client->header_or_short_response = response; + } // Switch states. - client->state = Client::SENDING_ERROR; + client->state = Client::SENDING_SHORT_RESPONSE; epoll_event ev; ev.events = EPOLLOUT | EPOLLET | EPOLLRDHUP; @@ -718,6 +986,10 @@ void Server::close_client(Client *client) delete_from(&client->stream->to_process, client); } + if (client->tls_context) { + tls_destroy_context(client->tls_context); + } + // Log to access_log. access_log->write(client->get_stats()); @@ -733,7 +1005,7 @@ void Server::process_queued_data() MutexLock lock(&queued_clients_mutex); for (size_t i = 0; i < queued_add_clients.size(); ++i) { - add_client(queued_add_clients[i]); + add_client(queued_add_clients[i].first, queued_add_clients[i].second); } queued_add_clients.clear(); }