client.cpp

#include <boost/asio/ip/tcp.hpp>
#include <boost/asio/spawn.hpp>
#include <boost/asio/connect.hpp>
#include <boost/asio/signal_set.hpp>
#include <boost/beast/core.hpp>
#include <boost/beast/http.hpp>
#include <boost/beast/version.hpp>
#include <boost/date_time/posix_time/posix_time.hpp>
#include <boost/format.hpp>
#include <boost/asio/ssl.hpp>
#include <boost/asio/ssl/stream.hpp>
#include <boost/optional/optional_io.hpp>
#include <boost/range/adaptor/indirected.hpp>
#include <boost/range/adaptor/filtered.hpp>
#include <boost/range/adaptor/transformed.hpp>
#include <boost/range/adaptor/indexed.hpp>
#include <iostream>
#include <cstdlib>  // for atexit()

#include "cache/bep5_http/client.h"

#include "namespaces.h"
#include "origin_pools.h"
#include "http_util.h"
#include "fetch_http_page.h"
#include "client_front_end.h"
#include "generic_stream.h"
#include "util.h"
#include "async_sleep.h"
#include "endpoint.h"
#include "or_throw.h"
#include "request_routing.h"
#include "full_duplex_forward.h"
#include "client_config.h"
#include "client.h"
#include "authenticate.h"
#include "defer.h"
#include "default_timeout.h"
#include "constants.h"
#include "util/async_queue_reader.h"
#include "session.h"
#include "create_udp_multiplexer.h"
#include "ssl/ca_certificate.h"
#include "ssl/dummy_certificate.h"
#include "ssl/util.h"
#include "bittorrent/dht.h"
#include "bittorrent/mutable_data.h"

#ifndef __ANDROID__
#  include "force_exit_on_signal.h"
#endif // ifndef __ANDROID__

#include "ouiservice.h"
#include "ouiservice/i2p.h"
#include "ouiservice/lampshade.h"
#include "ouiservice/pt-obfs2.h"
#include "ouiservice/pt-obfs3.h"
#include "ouiservice/pt-obfs4.h"
#include "ouiservice/tcp.h"
#include "ouiservice/utp.h"
#include "ouiservice/tls.h"
#include "ouiservice/weak_client.h"
#include "ouiservice/bep5/client.h"
#include "ouiservice/multi_utp_server.h"

#include "parse/number.h"
#include "util/signal.h"
#include "util/crypto.h"
#include "util/lru_cache.h"
#include "util/scheduler.h"
#include "util/reachability.h"
#include "util/async_job.h"
#include "upnp.h"
#include "util/handler_tracker.h"

#include "logger.h"

using namespace std;
using namespace ouinet;

namespace posix_time = boost::posix_time;
namespace bt = ouinet::bittorrent;

using tcp      = asio::ip::tcp;
using Request  = http::request<http::string_body>;
using Response = http::response<http::dynamic_body>;

static const fs::path OUINET_CA_CERT_FILE = "ssl-ca-cert.pem";
static const fs::path OUINET_CA_KEY_FILE = "ssl-ca-key.pem";
static const fs::path OUINET_CA_DH_FILE = "ssl-ca-dh.pem";

static bool log_transactions() {
    return logger.get_threshold() <= DEBUG
        || logger.get_log_file() != nullptr;
}

//------------------------------------------------------------------------------
struct UserAgentMetaData {
    boost::optional<bool> is_private;
    boost::optional<std::string> dht_group;

    static UserAgentMetaData extract(Request& rq) {
        UserAgentMetaData ret;

        {
            auto i = rq.find(http_::request_group_hdr);
            if (i != rq.end()) {
                ret.dht_group = i->value().to_string();
                rq.erase(i);
            }
        }
        {
            auto i = rq.find(http_::request_private_hdr);
            if (i != rq.end()) {
                ret.is_private = boost::iequals(i->value(), http_::request_private_true);
                rq.erase(i);
            }
        }

        return ret;
    }
};

//------------------------------------------------------------------------------
class Client::State : public enable_shared_from_this<Client::State> {
    friend class Client;

public:
    State(asio::io_context& ctx, ClientConfig cfg)
        : _ctx(ctx)
        , _config(move(cfg))
        // A certificate chain with OUINET_CA + SUBJECT_CERT
        // can be around 2 KiB, so this would be around 2 MiB.
        // TODO: Fine tune if necessary.
        , _ssl_certificate_cache(1000)
        , ssl_ctx{asio::ssl::context::tls_client}
        , inj_ctx{asio::ssl::context::tls_client}
    {
        ssl_ctx.set_default_verify_paths();
        ssl_ctx.set_verify_mode(asio::ssl::verify_peer);

        // We do *not* want to do this since
        // we will not be checking certificate names,
        // thus any certificate signed by a recognized CA
        // would be accepted if presented by an injector.
        //inj_ctx.set_default_verify_paths();
        inj_ctx.set_verify_mode(asio::ssl::verify_peer);
    }

    void start();

    void stop() {
        _bep5_http_cache = nullptr;
        _upnps.clear();
        _shutdown_signal();
        if (_injector) _injector->stop();
        if (_bt_dht) {
            _bt_dht->stop();
            _bt_dht = nullptr;
        }
        if (_udp_reachability) {
            _udp_reachability->stop();
            _udp_reachability = nullptr;
        }
    }

    void setup_cache();
    void set_injector(string);

    const asio_utp::udp_multiplexer& common_udp_multiplexer()
    {
        if (_udp_multiplexer) return *_udp_multiplexer;

        _udp_multiplexer
            = create_udp_multiplexer( _ctx
                                    , _config.repo_root() / "last_used_udp_port");

        _udp_reachability
            = make_unique<util::UdpServerReachabilityAnalysis>();
        _udp_reachability->start(get_executor(), *_udp_multiplexer);

        return *_udp_multiplexer;
    }

    std::shared_ptr<bt::MainlineDht> bittorrent_dht(asio::yield_context yield)
    {
        if (_bt_dht) return _bt_dht;

        auto bt_dht = make_shared<bt::MainlineDht>( _ctx.get_executor()
                                                  , _config.repo_root() / "dht");

        auto& mpl = common_udp_multiplexer();

        asio_utp::udp_multiplexer m(_ctx);
        sys::error_code ec;

        m.bind(mpl, ec);
        if (ec) return or_throw(yield, ec, _bt_dht);

        auto cc = _shutdown_signal.connect([&] { bt_dht.reset(); });

        auto ext_ep = bt_dht->set_endpoint(move(m), yield[ec]);
        if (ec) return or_throw(yield, ec, _bt_dht);

        setup_upnp(ext_ep.port(), mpl.local_endpoint());

        _bt_dht = move(bt_dht);
        return _bt_dht;
    }

    http::response<http::string_body>
    retrieval_failure_response(const Request&);

private:
    GenericStream ssl_mitm_handshake( GenericStream&&
                                    , const Request&
                                    , asio::yield_context);

    void serve_request(GenericStream&& con, asio::yield_context yield);

    // All `fetch_*` functions below take care of keeping or dropping
    // Ouinet-specific internal HTTP headers as expected by upper layers.

    CacheEntry
    fetch_stored_in_dcache( const Request& request
                          , const request_route::Config& request_config
                          , const std::string& dht_group
                          , Cancel& cancel
                          , Yield yield);

    Response fetch_fresh_from_front_end(const Request&, Yield);
    Session fetch_fresh_from_origin(const Request&, Cancel, Yield);

    Session fetch_fresh_through_connect_proxy(const Request&, Cancel&, Yield);

    Session fetch_fresh_through_simple_proxy( Request
                                            , bool can_inject
                                            , Cancel& cancel
                                            , Yield);

    template<class Resp>
    void maybe_add_proto_version_warning(Resp& res) const {
        auto newest = newest_proto_seen;
        // Check if cache client knows about a newer protocol version too.
        auto c = get_cache();
        if (c && c->get_newest_proto_version() > newest)
            newest = c->get_newest_proto_version();
        if (newest > http_::protocol_version_current)
            res.set( http_::response_warning_hdr
                   , "Newer Ouinet protocol found in network, "
                     "please consider upgrading.");
    };

    CacheControl build_cache_control(request_route::Config& request_config);

    void listen_tcp( asio::yield_context
                   , tcp::endpoint
                   , const char* service
                   , function<void(GenericStream, asio::yield_context)>);

    void setup_injector(asio::yield_context);

    bool was_stopped() const {
        return _shutdown_signal.call_count() != 0;
    }

    fs::path ca_cert_path() const { return _config.repo_root() / OUINET_CA_CERT_FILE; }
    fs::path ca_key_path()  const { return _config.repo_root() / OUINET_CA_KEY_FILE;  }
    fs::path ca_dh_path()   const { return _config.repo_root() / OUINET_CA_DH_FILE;   }

    asio::io_context& get_io_context() { return _ctx; }
    asio::executor get_executor() { return _ctx.get_executor(); }

    Signal<void()>& get_shutdown_signal() { return _shutdown_signal; }

    bool maybe_handle_websocket_upgrade( GenericStream&
                                       , beast::string_view connect_host_port
                                       , Request&
                                       , Yield);

    GenericStream connect_to_origin(const Request&, Cancel&, Yield);

    unique_ptr<OuiServiceImplementationClient>
    maybe_wrap_tls(unique_ptr<OuiServiceImplementationClient>);

    cache::bep5_http::Client* get_cache() const { return _bep5_http_cache.get(); }

    void serve_utp_request(GenericStream, Yield);

    void setup_upnp(uint16_t ext_port, asio::ip::udp::endpoint local_ep) {
        if (_shutdown_signal) return;

        if (!local_ep.address().is_v4()) {
            LOG_WARN("Not setting up UPnP redirection because endpoint is not ipv4");
            return;
        }

        auto& p = _upnps[local_ep];

        if (p) {
            LOG_WARN("UPnP redirection for ", local_ep, " is already set");
            return;
        }

        p = make_unique<UPnPUpdater>(_ctx.get_executor(), ext_port, local_ep.port());
    }

    void idempotent_start_accepting_on_utp(asio::yield_context yield) {
        if (_multi_utp_server) return;
        assert(!_shutdown_signal);
        if (_shutdown_signal) return or_throw(yield, asio::error::operation_aborted);

        sys::error_code ec;
        auto dht = bittorrent_dht(yield[ec]);
        if (ec) return or_throw(yield, ec);

        auto exec = _ctx.get_executor();
        auto local_eps = dht->local_endpoints();

        _multi_utp_server
            = make_unique<ouiservice::MultiUtpServer>(exec, local_eps, nullptr);

        TRACK_SPAWN(_ctx, ([&, c = _shutdown_signal] (asio::yield_context yield) mutable {
            auto slot = c.connect([&] () mutable { _multi_utp_server = nullptr; });

            sys::error_code ec;
            _multi_utp_server->start_listen(yield[ec]);

            if (ec) {
                LOG_ERROR("Failed to start accepting on multi uTP service: ", ec.message());
                return;
            }

            while (!c) {
                sys::error_code ec;
                auto con = _multi_utp_server->accept(yield[ec]);
                if (c) return;
                if (ec == asio::error::operation_aborted) return;
                if (ec) {
                    LOG_WARN("Bep5Http: Failure to accept:", ec.message());
                    async_sleep(_ctx, 200ms, c, yield);
                    continue;
                }
                TRACK_SPAWN(_ctx, ([&, con = move(con)]
                                   (asio::yield_context yield) mutable {
                    Yield y(_ctx, yield, "uTPAccept");
                    serve_utp_request(move(con), y[ec]);
                }));
            }
        }));
    }

private:
    // The newest protocol version number seen in a trusted exchange
    // (i.e. from an injector exchange or injector-signed cached content).
    unsigned newest_proto_seen = http_::protocol_version_current;

    asio::io_context& _ctx;
    ClientConfig _config;
    std::unique_ptr<CACertificate> _ca_certificate;
    util::LruCache<string, string> _ssl_certificate_cache;
    std::unique_ptr<OuiServiceClient> _injector;
    std::unique_ptr<cache::bep5_http::Client> _bep5_http_cache;

    ClientFrontEnd _front_end;
    Signal<void()> _shutdown_signal;

    // For debugging
    uint64_t _next_connection_id = 0;
    ConnectionPool<Endpoint> _injector_connections;
    OriginPools _origin_pools;

    asio::ssl::context ssl_ctx;
    asio::ssl::context inj_ctx;

    boost::optional<asio::ip::udp::endpoint> _local_utp_endpoint;
    boost::optional<asio_utp::udp_multiplexer> _udp_multiplexer;
    unique_ptr<util::UdpServerReachabilityAnalysis> _udp_reachability;
    shared_ptr<bt::MainlineDht> _bt_dht;

    unique_ptr<ouiservice::MultiUtpServer> _multi_utp_server;
    shared_ptr<ouiservice::Bep5Client> _bep5_client;

    std::map<asio::ip::udp::endpoint, unique_ptr<UPnPUpdater>> _upnps;
};

//------------------------------------------------------------------------------
template<class Resp>
static
void handle_http_error( GenericStream& con
                      , Resp& res
                      , Yield yield)
{
    if (log_transactions()) {
        yield.log("=== Sending back response ===");
        yield.log(res);
    }

    http::async_write(con, res, yield);
}

template<class ReqBody>
static
void handle_bad_request( GenericStream& con
                       , const http::request<ReqBody>& req
                       , const string& message
                       , Yield yield)
{
    auto res = util::http_client_error(req, http::status::bad_request, "", message);
    auto yield_ = yield.tag("handle_bad_request");
    return handle_http_error(con, res, yield_);
}

//------------------------------------------------------------------------------
void
Client::State::serve_utp_request(GenericStream con, Yield yield)
{
    assert(_bep5_http_cache);
    Cancel cancel = _shutdown_signal;

    sys::error_code ec;

    http::request<http::empty_body> req;
    beast::flat_buffer buffer;

    {
        WatchDog watch_dog(_ctx , chrono::seconds(5) , [&] { con.close(); });

        http::async_read(con, buffer, req, yield[ec].tag("read"));

        if (!watch_dog.is_running()) {
            return or_throw(yield, asio::error::timed_out);
        }
    }

    if (ec || cancel) return;

    if (req.method() != http::verb::connect) {
        _bep5_http_cache->serve_local(req, con, cancel, yield[ec]);
        return;
    }

    // Connect to the injector and tunnel the transaction through it

    if (!_bep5_client) {
        return handle_bad_request(con, req, "No known injectors", yield[ec]);
    }

    auto inj = _bep5_client->connect( yield[ec].tag("connect_to_injector")
                                    , cancel
                                    , false
                                    , ouiservice::Bep5Client::injectors);

    if (cancel) ec = asio::error::operation_aborted;
    if (ec == asio::error::operation_aborted) return;
    if (ec) {
        ec = {};
        return handle_bad_request(con, req, "Failed to connect to injector", yield[ec]);
    }

    // Send the client an OK message indicating that the tunnel
    // has been established.
    http::response<http::empty_body> res{http::status::ok, req.version()};
    res.prepare_payload();

    // No ``res.prepare_payload()`` since no payload is allowed for CONNECT:
    // <https://tools.ietf.org/html/rfc7231#section-6.3.1>.

    http::async_write(con, res, yield[ec].tag("write"));

    if (cancel) ec = asio::error::operation_aborted;
    if (ec) return;

    full_duplex(move(con), move(inj), yield[ec].tag("full_duplex"));
}

//------------------------------------------------------------------------------
CacheEntry
Client::State::fetch_stored_in_dcache( const Request& request
                                     , const request_route::Config& request_config
                                     , const std::string& dht_group
                                     , Cancel& cancel
                                     , Yield yield)
{
    auto c = get_cache();

    const bool cache_is_disabled
        = !c
       || !_config.is_cache_access_enabled();

    if (cache_is_disabled) {
        return or_throw<CacheEntry>( yield
                                   , asio::error::operation_not_supported);
    }

    sys::error_code ec;

    auto key = key_from_http_req(request);
    assert(key);
    auto s = c->load(move(*key), dht_group, cancel, yield[ec]);
    return_or_throw_on_error(yield, cancel, ec, CacheEntry{});

    auto& hdr = s.response_header();

    if (!util::http_proto_version_check_trusted(hdr, newest_proto_seen))
        // The cached resource cannot be used, treat it like
        // not being found.
        return or_throw<CacheEntry>(yield, asio::error::not_found);

    auto tsh = util::http_injection_ts(hdr);
    auto ts = parse::number<time_t>(tsh);
    auto date = ( ts
                ? boost::posix_time::from_time_t(*ts)
                : boost::posix_time::not_a_date_time);

    maybe_add_proto_version_warning(hdr);
    assert(!hdr[http_::response_source_hdr].empty());  // for agent, set by cache
    return CacheEntry{date, move(s)};
}

//------------------------------------------------------------------------------
GenericStream
Client::State::connect_to_origin( const Request& rq
                                , Cancel& cancel
                                , Yield yield)
{
    std::string host, port;
    std::tie(host, port) = util::get_host_port(rq);

    sys::error_code ec;

    auto lookup = util::tcp_async_resolve( host, port
                                         , _ctx.get_executor()
                                         , cancel
                                         , yield[ec]);

    return_or_throw_on_error(yield, cancel, ec, GenericStream());

    auto sock = connect_to_host(lookup, _ctx.get_executor(), cancel, yield[ec]);

    return_or_throw_on_error(yield, cancel, ec, GenericStream());

    GenericStream stream;

    if (rq.target().starts_with("https:") || rq.target().starts_with("wss:")) {
        stream = ssl::util::client_handshake( move(sock)
                                            , ssl_ctx
                                            , host
                                            , cancel
                                            , yield[ec]);

        return_or_throw_on_error(yield, cancel, ec, GenericStream());
    }
    else {
        stream = move(sock);
    }

    return stream;
}
//------------------------------------------------------------------------------
Response Client::State::fetch_fresh_from_front_end(const Request& rq, Yield yield)
{
    boost::optional<uint32_t> udp_port;

    if (_udp_multiplexer) {
        udp_port = _udp_multiplexer->local_endpoint().port();
    }

    auto res = _front_end.serve( _config
                               , rq
                               , _bep5_http_cache.get()
                               , *_ca_certificate
                               , udp_port
                               , _upnps
                               , _udp_reachability.get()
                               , yield.tag("serve_frontend"));

    res.set( http_::response_source_hdr  // for agent
           , http_::response_source_hdr_front_end);

    res.keep_alive(rq.keep_alive());

    return res;
}

//------------------------------------------------------------------------------
Session Client::State::fetch_fresh_from_origin(const Request& rq, Cancel cancel, Yield yield)
{
    WatchDog watch_dog(_ctx
                      , default_timeout::fetch_http()
                      , [&] { cancel(); });

    sys::error_code ec;

    auto maybe_con = _origin_pools.get_connection(rq);
    OriginPools::Connection con;
    if (maybe_con) {
        con = std::move(*maybe_con);
    } else {
        auto stream = connect_to_origin(rq, cancel, yield[ec]);

        if (cancel) {
            assert(ec == asio::error::operation_aborted);
            ec = watch_dog.is_running()
               ? ec = asio::error::operation_aborted
               : ec = asio::error::timed_out;
        }

        if (ec) return or_throw<Session>(yield, ec);

        con = _origin_pools.wrap(rq, std::move(stream));
    }

    // Transform request from absolute-form to origin-form
    // https://tools.ietf.org/html/rfc7230#section-5.3
    auto rq_ = util::req_form_from_absolute_to_origin(rq);

    // Send request
    {
        auto con_close = cancel.connect([&] { con.close(); });
        http::async_write(con, rq_, yield[ec].tag("send-origin-request"));
    }

    if (cancel) {
        ec = watch_dog.is_running() ? ec = asio::error::operation_aborted
                                    : ec = asio::error::timed_out;
    }
    if (ec) return or_throw<Session>(yield, ec);

    auto ret = Session::create(std::move(con), cancel, yield[ec]);
    return_or_throw_on_error(yield, cancel, ec, Session());

    // Prevent others from inserting ouinet headers.
    util::remove_ouinet_fields_ref(ret.response_header());

    ret.response_header().set( http_::response_source_hdr  // for agent
                             , http_::response_source_hdr_origin);
    return ret;
}

//------------------------------------------------------------------------------
Session Client::State::fetch_fresh_through_connect_proxy( const Request& rq
                                                        , Cancel& cancel_
                                                        , Yield yield)
{
    // TODO: We're not re-using connections here. It's because the
    // ConnectionPool as it is right now can only work with http requests
    // and responses and thus can't be used for full-dupplex forwarding.

    Cancel cancel(cancel_);
    WatchDog watch_dog(_ctx, default_timeout::fetch_http(), [&]{ cancel(); });

    // Parse the URL to tell HTTP/HTTPS, host, port.
    util::url_match url;

    if (!match_http_url(rq.target(), url)) {
        // unsupported URL
        return or_throw<Session>(yield, asio::error::operation_not_supported);
    }

    // Connect to the injector/proxy.
    sys::error_code ec;

    auto inj = _injector->connect(yield[ec].tag("connect_to_injector"), cancel);

    if (ec) return or_throw<Session>(yield, ec);

    // Build the actual request to send to the proxy.
    Request connreq = { http::verb::connect
                      , url.host + ":" + (url.port.empty() ? "443" : url.port)
                      , 11 /* HTTP/1.1 */};

    // HTTP/1.1 requires a ``Host:`` header in all requests:
    // <https://tools.ietf.org/html/rfc7230#section-5.4>.
    connreq.set(http::field::host, connreq.target());

    if (auto credentials = _config.credentials_for(inj.remote_endpoint))
        connreq = authorize(connreq, *credentials);

    // Open a tunnel to the origin
    // (to later perform the SSL handshake and send the request).
    // Only get the head of the CONNECT response
    // (otherwise we would get stuck waiting to read
    // a body whose length we do not know
    // since the respone should have no content length).
    auto connres = fetch_http<http::empty_body>
                        ( _ctx.get_executor()
                        , inj.connection
                        , connreq
                        , default_timeout::fetch_http()
                        , cancel
                        , yield[ec].tag("connreq"));

    if (connres.result() != http::status::ok) {
        // This error code is quite fake, so log the error too.
        // Unfortunately there is no body to show.
        yield.tag("proxy_connect").log(connres);
        return or_throw<Session>(yield, asio::error::connection_refused);
    }

    GenericStream con;

    if (url.scheme == "https") {
        con = ssl::util::client_handshake( move(inj.connection)
                                         , ssl_ctx
                                         , url.host
                                         , cancel
                                         , yield[ec]);
    } else {
        con = move(inj.connection);
    }

    if (ec) return or_throw<Session>(yield, ec);

    // TODO: move
    auto rq_ = util::req_form_from_absolute_to_origin(rq);

    {
        auto slot = cancel.connect([&con] { con.close(); });
        http::async_write(con, rq_, yield[ec]);
        return_or_throw_on_error(yield, cancel, ec, Session());
    }

    auto session = Session::create(move(con), cancel, yield[ec]);
    return_or_throw_on_error(yield, cancel, ec, Session());

    // Prevent others from inserting ouinet headers.
    util::remove_ouinet_fields_ref(session.response_header());

    session.response_header().set( http_::response_source_hdr  // for agent
                                 , http_::response_source_hdr_proxy);
    return session;
}
//------------------------------------------------------------------------------
Session Client::State::fetch_fresh_through_simple_proxy
        ( Request request
        , bool can_inject
        , Cancel& cancel
        , Yield yield)
{
    sys::error_code ec;

    // Connect to the injector.
    ConnectionPool<Endpoint>::Connection con;
    if (_injector_connections.empty()) {
        if (log_transactions()) {
            yield.log("Connecting to the injector");
        }

        auto c = _injector->connect(yield[ec].tag("connect_to_injector2"), cancel);

        assert(!cancel || ec == asio::error::operation_aborted);

        if (ec) {
            if (log_transactions()) {
                yield.log("Failed to connect to injector ec:", ec.message());
            }
            return or_throw<Session>(yield, ec);
        }

        assert(c.connection.has_implementation());

        con = _injector_connections.wrap(std::move(c.connection));
        *con = c.remote_endpoint;
    } else {
        if (log_transactions()) {
            yield.log("Reusing existing injector connection");
        }

        con = _injector_connections.pop_front();
    }

    auto cancel_slot = cancel.connect([&] {
        con.close();
    });

    // Build the actual request to send to the injector.
    if (auto credentials = _config.credentials_for(*con))
        request = authorize(request, *credentials);

    if (can_inject) {
        bool keepalive = request.keep_alive();
        request = util::to_injector_request(move(request));
        request.keep_alive(keepalive);
    }

    if (log_transactions()) {
        yield.log("Sending a request to the injector");
    }
    // Send request
    http::async_write(con, request, yield[ec].tag("inj-request"));

    if (!ec && cancel_slot) {
        ec = asio::error::operation_aborted;
    }

    if (ec && log_transactions()) {
        yield.log("Failed to send request to the injector");
    }

    if (ec) return or_throw<Session>(yield, ec);

    if (log_transactions()) {
        yield.log("Reading response");
    }

    cancel_slot = {};

    // Receive response
    auto session = Session::create(move(con), cancel, yield[ec]);

    auto& hdr = session.response_header();

    if (cancel)
        ec = asio::error::operation_aborted;
    else if ( !ec
            && can_inject
            && !util::http_proto_version_check_trusted(hdr, newest_proto_seen))
        // The injector using an unacceptable protocol version is treated like
        // the Injector mechanism being disabled.
        ec = asio::error::operation_not_supported;

    if (log_transactions()) {
        yield.log("End reading response. ec:", ec.message());
    }

    if (ec) return or_throw(yield, ec, std::move(session));

    // Store keep-alive connections in connection pool

    if (can_inject) {
        maybe_add_proto_version_warning(hdr);

        session.response_header().set( http_::response_source_hdr  // for agent
                                     , http_::response_source_hdr_injector);
    } else {
        // Prevent others from inserting ouinet headers.
        util::remove_ouinet_fields_ref(hdr);

        session.response_header().set( http_::response_source_hdr  // for agent
                                     , http_::response_source_hdr_proxy);
    }
    return session;
}

class Transaction {
public:
    Transaction(GenericStream& ua_con, const Request& rq, UserAgentMetaData meta)
        : _ua_con(ua_con)
        , _request(rq)
        , _meta(std::move(meta))
    {}

    void write_to_user_agent(Session& session, Cancel& cancel, asio::yield_context yield)
    {
        namespace err = asio::error;

        if (cancel) {
            assert(!cancel);
            LOG_ERROR(__FILE__, ":", __LINE__, " Cancel already called");
            return or_throw(yield, err::operation_aborted);
        }

        if (_ua_was_written_to) {
            return or_throw(yield, err::already_started);
        }

        sys::error_code ec;

        _ua_was_written_to = true;
        session.flush_response(_ua_con, cancel, yield[ec]);

        bool keep_alive = !ec && _request.keep_alive() && session.keep_alive();

        if (!keep_alive) {
            session.close();
            _ua_con.close();
        }

        return or_throw(yield, ec);
    }

    template<class BodyType>
    void write_to_user_agent(const http::response<BodyType>& rs, Cancel& cancel, asio::yield_context yield)
    {
        namespace err = asio::error;