src/clients/HttpTunneler.cc

   1 /*
   2  * Copyright (C) 1996-2020 The Squid Software Foundation and contributors
   3  *
   4  * Squid software is distributed under GPLv2+ license and includes
   5  * contributions from numerous individuals and organizations.
   6  * Please see the COPYING and CONTRIBUTORS files for details.
   7  */
   8
   9 #include "squid.h"
  10 #include "CachePeer.h"
  11 #include "clients/HttpTunneler.h"
  12 #include "comm/Read.h"
  13 #include "comm/Write.h"
  14 #include "errorpage.h"
  15 #include "fd.h"
  16 #include "fde.h"
  17 #include "http.h"
  18 #include "http/one/ResponseParser.h"
  19 #include "http/StateFlags.h"
  20 #include "HttpRequest.h"
  21 #include "neighbors.h"
  22 #include "pconn.h"
  23 #include "SquidConfig.h"
  24 #include "StatCounters.h"
  25
  26 CBDATA_NAMESPACED_CLASS_INIT(Http, Tunneler);
  27
  28 Http::Tunneler::Tunneler(const Comm::ConnectionPointer &conn, const HttpRequest::Pointer &req, AsyncCall::Pointer &aCallback, time_t timeout, const AccessLogEntryPointer &alp):
  29     AsyncJob("Http::Tunneler"),
  30     noteFwdPconnUse(false),
  31     connection(conn),
  32     request(req),
  33     callback(aCallback),
  34     lifetimeLimit(timeout),
  35     al(alp),
  36     startTime(squid_curtime),
  37     requestWritten(false),
  38     tunnelEstablished(false)
  39 {
  40     debugs(83, 5, "Http::Tunneler constructed, this=" << (void*)this);
  41     // detect callers supplying cb dialers that are not our CbDialer
  42     assert(request);
  43     assert(connection);
  44     assert(callback);
  45     assert(dynamic_cast<Http::TunnelerAnswer *>(callback->getDialer()));
  46     url = request->url.authority();
  47     watchForClosures();
  48 }
  49
  50 Http::Tunneler::~Tunneler()
  51 {
  52     debugs(83, 5, "Http::Tunneler destructed, this=" << (void*)this);
  53 }
  54
  55 bool
  56 Http::Tunneler::doneAll() const
  57 {
  58     return !callback || (requestWritten && tunnelEstablished);
  59 }
  60
  61 /// convenience method to get to the answer fields
  62 Http::TunnelerAnswer &
  63 Http::Tunneler::answer()
  64 {
  65     Must(callback);
  66     const auto tunnelerAnswer = dynamic_cast<Http::TunnelerAnswer *>(callback->getDialer());
  67     Must(tunnelerAnswer);
  68     return *tunnelerAnswer;
  69 }
  70
  71 void
  72 Http::Tunneler::start()
  73 {
  74     AsyncJob::start();
  75
  76     Must(al);
  77     Must(url.length());
  78     Must(lifetimeLimit >= 0);
  79
  80     // we own this Comm::Connection object and its fd exclusively, but must bail
  81     // if others started closing the socket while we were waiting to start()
  82     assert(Comm::IsConnOpen(connection));
  83     if (fd_table[connection->fd].closing()) {
  84         bailWith(new ErrorState(ERR_CONNECT_FAIL, Http::scBadGateway, request.getRaw(), al));
  85         return;
  86     }
  87
  88     const auto peer = connection->getPeer();
  89     // bail if our peer was reconfigured away
  90     if (!peer) {
  91         bailWith(new ErrorState(ERR_CONNECT_FAIL, Http::scInternalServerError, request.getRaw(), al));
  92         return;
  93     }
  94     request->prepForPeering(*peer);
  95
  96     writeRequest();
  97     startReadingResponse();
  98 }
  99
 100 void
 101 Http::Tunneler::handleConnectionClosure(const CommCloseCbParams &params)
 102 {
 103     closer = nullptr;
 104     bailWith(new ErrorState(ERR_CONNECT_FAIL, Http::scBadGateway, request.getRaw(), al));
 105 }
 106
 107 /// make sure we quit if/when the connection is gone
 108 void
 109 Http::Tunneler::watchForClosures()
 110 {
 111     Must(Comm::IsConnOpen(connection));
 112     Must(!fd_table[connection->fd].closing());
 113
 114     debugs(83, 5, connection);
 115
 116     Must(!closer);
 117     typedef CommCbMemFunT<Http::Tunneler, CommCloseCbParams> Dialer;
 118     closer = JobCallback(9, 5, Dialer, this, Http::Tunneler::handleConnectionClosure);
 119     comm_add_close_handler(connection->fd, closer);
 120 }
 121
 122 /// The connection read timeout callback handler.
 123 void
 124 Http::Tunneler::handleTimeout(const CommTimeoutCbParams &)
 125 {
 126     bailWith(new ErrorState(ERR_CONNECT_FAIL, Http::scGatewayTimeout, request.getRaw(), al));
 127 }
 128
 129 void
 130 Http::Tunneler::startReadingResponse()
 131 {
 132     debugs(83, 5, connection << status());
 133
 134     readBuf.reserveCapacity(SQUID_TCP_SO_RCVBUF);
 135     readMore();
 136 }
 137
 138 void
 139 Http::Tunneler::writeRequest()
 140 {
 141     debugs(83, 5, connection);
 142
 143     Http::StateFlags flags;
 144     flags.peering = true;
 145     // flags.tunneling = false; // the CONNECT request itself is not tunneled
 146     // flags.toOrigin = false; // the next HTTP hop is a non-originserver peer
 147
 148     MemBuf mb;
 149
 150     try {
 151         request->masterXaction->generatingConnect = true;
 152
 153         mb.init();
 154         mb.appendf("CONNECT %s HTTP/1.1\r\n", url.c_str());
 155         HttpHeader hdr_out(hoRequest);
 156         HttpStateData::httpBuildRequestHeader(request.getRaw(),
 157                                               nullptr, // StoreEntry
 158                                               al,
 159                                               &hdr_out,
 160                                               flags);
 161         hdr_out.packInto(&mb);
 162         hdr_out.clean();
 163         mb.append("\r\n", 2);
 164
 165         request->masterXaction->generatingConnect = false;
 166     } catch (...) {
 167         // TODO: Add scope_guard; do not wait until it is in the C++ standard.
 168         request->masterXaction->generatingConnect = false;
 169         throw;
 170     }
 171
 172     debugs(11, 2, "Tunnel Server REQUEST: " << connection <<
 173            ":\n----------\n" << mb.buf << "\n----------");
 174     fd_note(connection->fd, "Tunnel Server CONNECT");
 175
 176     typedef CommCbMemFunT<Http::Tunneler, CommIoCbParams> Dialer;
 177     writer = JobCallback(5, 5, Dialer, this, Http::Tunneler::handleWrittenRequest);
 178     Comm::Write(connection, &mb, writer);
 179 }
 180
 181 /// Called when we are done writing a CONNECT request header to a peer.
 182 void
 183 Http::Tunneler::handleWrittenRequest(const CommIoCbParams &io)
 184 {
 185     Must(writer);
 186     writer = nullptr;
 187
 188     if (io.flag == Comm::ERR_CLOSING)
 189         return;
 190
 191     request->hier.notePeerWrite();
 192
 193     if (io.flag != Comm::OK) {
 194         const auto error = new ErrorState(ERR_WRITE_ERROR, Http::scBadGateway, request.getRaw(), al);
 195         error->xerrno = io.xerrno;
 196         bailWith(error);
 197         return;
 198     }
 199
 200     statCounter.server.all.kbytes_out += io.size;
 201     statCounter.server.other.kbytes_out += io.size;
 202     requestWritten = true;
 203     debugs(83, 5, status());
 204 }
 205
 206 /// Called when we read [a part of] CONNECT response from the peer
 207 void
 208 Http::Tunneler::handleReadyRead(const CommIoCbParams &io)
 209 {
 210     Must(reader);
 211     reader = nullptr;
 212
 213     if (io.flag == Comm::ERR_CLOSING)
 214         return;
 215
 216     CommIoCbParams rd(this);
 217     rd.conn = io.conn;
 218 #if USE_DELAY_POOLS
 219     rd.size = delayId.bytesWanted(1, readBuf.spaceSize());
 220 #else
 221     rd.size = readBuf.spaceSize();
 222 #endif
 223     // XXX: defer read if rd.size <= 0
 224
 225     switch (Comm::ReadNow(rd, readBuf)) {
 226     case Comm::INPROGRESS:
 227         readMore();
 228         return;
 229
 230     case Comm::OK: {
 231 #if USE_DELAY_POOLS
 232         delayId.bytesIn(rd.size);
 233 #endif
 234         statCounter.server.all.kbytes_in += rd.size;
 235         statCounter.server.other.kbytes_in += rd.size; // TODO: other or http?
 236         request->hier.notePeerRead();
 237         handleResponse(false);
 238         return;
 239     }
 240
 241     case Comm::ENDFILE: {
 242         // TODO: Should we (and everybody else) call request->hier.notePeerRead() on zero reads?
 243         handleResponse(true);
 244         return;
 245     }
 246
 247     // case Comm::COMM_ERROR:
 248     default: // no other flags should ever occur
 249     {
 250         const auto error = new ErrorState(ERR_READ_ERROR, Http::scBadGateway, request.getRaw(), al);
 251         error->xerrno = rd.xerrno;
 252         bailWith(error);
 253         return;
 254     }
 255     }
 256
 257     assert(false); // not reached
 258 }
 259
 260 void
 261 Http::Tunneler::readMore()
 262 {
 263     Must(Comm::IsConnOpen(connection));
 264     Must(!fd_table[connection->fd].closing());
 265     Must(!reader);
 266
 267     typedef CommCbMemFunT<Http::Tunneler, CommIoCbParams> Dialer;
 268     reader = JobCallback(93, 3, Dialer, this, Http::Tunneler::handleReadyRead);
 269     Comm::Read(connection, reader);
 270
 271     AsyncCall::Pointer nil;
 272     typedef CommCbMemFunT<Http::Tunneler, CommTimeoutCbParams> TimeoutDialer;
 273     AsyncCall::Pointer timeoutCall = JobCallback(93, 5,
 274                                      TimeoutDialer, this, Http::Tunneler::handleTimeout);
 275     const auto timeout = Comm::MortalReadTimeout(startTime, lifetimeLimit);
 276     commSetConnTimeout(connection, timeout, timeoutCall);
 277 }
 278
 279 /// Parses [possibly incomplete] CONNECT response and reacts to it.
 280 void
 281 Http::Tunneler::handleResponse(const bool eof)
 282 {
 283     // mimic the basic parts of HttpStateData::processReplyHeader()
 284     if (hp == nullptr)
 285         hp = new Http1::ResponseParser;
 286
 287     auto parsedOk = hp->parse(readBuf); // may be refined below
 288     readBuf = hp->remaining();
 289     if (hp->needsMoreData()) {
 290         if (!eof) {
 291             if (readBuf.length() >= SQUID_TCP_SO_RCVBUF) {
 292                 bailOnResponseError("huge CONNECT response from peer", nullptr);
 293                 return;
 294             }
 295             readMore();
 296             return;
 297         }
 298
 299         //eof, handle truncated response
 300         readBuf.append("\r\n\r\n", 4);
 301         parsedOk = hp->parse(readBuf);
 302         readBuf.clear();
 303     }
 304
 305     if (!parsedOk) {
 306         bailOnResponseError("malformed CONNECT response from peer", nullptr);
 307         return;
 308     }
 309
 310     HttpReply::Pointer rep = new HttpReply;
 311     rep->sources |= Http::Message::srcHttp;
 312     rep->sline.set(hp->messageProtocol(), hp->messageStatus());
 313     if (!rep->parseHeader(*hp) && rep->sline.status() == Http::scOkay) {
 314         bailOnResponseError("malformed CONNECT response from peer", nullptr);
 315         return;
 316     }
 317
 318     // CONNECT response was successfully parsed
 319     auto &futureAnswer = answer();
 320     futureAnswer.peerResponseStatus = rep->sline.status();
 321     request->hier.peer_reply_status = rep->sline.status();
 322
 323     debugs(11, 2, "Tunnel Server " << connection);
 324     debugs(11, 2, "Tunnel Server RESPONSE:\n---------\n" <<
 325            Raw(nullptr, readBuf.rawContent(), rep->hdr_sz).minLevel(2).gap(false) <<
 326            "----------");
 327
 328     // bail if we did not get an HTTP 200 (Connection Established) response
 329     if (rep->sline.status() != Http::scOkay) {
 330         // TODO: To reuse the connection, extract the whole error response.
 331         bailOnResponseError("unsupported CONNECT response status code", rep.getRaw());
 332         return;
 333     }
 334
 335     // preserve any bytes sent by the server after the CONNECT response
 336     futureAnswer.leftovers = readBuf;
 337
 338     tunnelEstablished = true;
 339     debugs(83, 5, status());
 340 }
 341
 342 void
 343 Http::Tunneler::bailOnResponseError(const char *error, HttpReply *errorReply)
 344 {
 345     debugs(83, 3, error << status());
 346
 347     ErrorState *err;
 348     if (errorReply) {
 349         err = new ErrorState(request.getRaw(), errorReply);
 350     } else {
 351         // with no reply suitable for relaying, answer with 502 (Bad Gateway)
 352         err = new ErrorState(ERR_CONNECT_FAIL, Http::scBadGateway, request.getRaw(), al);
 353     }
 354     bailWith(err);
 355 }
 356
 357 void
 358 Http::Tunneler::bailWith(ErrorState *error)
 359 {
 360     Must(error);
 361     answer().squidError = error;
 362
 363     if (const auto p = connection->getPeer())
 364         peerConnectFailed(p);
 365
 366     callBack();
 367     disconnect();
 368
 369     if (noteFwdPconnUse)
 370         fwdPconnPool->noteUses(fd_table[connection->fd].pconn.uses);
 371     // TODO: Reuse to-peer connections after a CONNECT error response.
 372     connection->close();
 373     connection = nullptr;
 374 }
 375
 376 void
 377 Http::Tunneler::sendSuccess()
 378 {
 379     assert(answer().positive());
 380     callBack();
 381     disconnect();
 382 }
 383
 384 void
 385 Http::Tunneler::disconnect()
 386 {
 387     if (closer) {
 388         comm_remove_close_handler(connection->fd, closer);
 389         closer = nullptr;
 390     }
 391
 392     if (reader) {
 393         Comm::ReadCancel(connection->fd, reader);
 394         reader = nullptr;
 395     }
 396
 397     // remove connection timeout handler
 398     commUnsetConnTimeout(connection);
 399 }
 400
 401 void
 402 Http::Tunneler::callBack()
 403 {
 404     debugs(83, 5, connection << status());
 405     if (answer().positive())
 406         answer().conn = connection;
 407     auto cb = callback;
 408     callback = nullptr;
 409     ScheduleCallHere(cb);
 410 }
 411
 412 void
 413 Http::Tunneler::swanSong()
 414 {
 415     AsyncJob::swanSong();
 416
 417     if (callback) {
 418         if (requestWritten && tunnelEstablished) {
 419             sendSuccess();
 420         } else {
 421             // we should have bailed when we discovered the job-killing problem
 422             debugs(83, DBG_IMPORTANT, "BUG: Unexpected state while establishing a CONNECT tunnel " << connection << status());
 423             bailWith(new ErrorState(ERR_GATEWAY_FAILURE, Http::scInternalServerError, request.getRaw(), al));
 424         }
 425         assert(!callback);
 426     }
 427 }
 428
 429 const char *
 430 Http::Tunneler::status() const
 431 {
 432     static MemBuf buf;
 433     buf.reset();
 434
 435     // TODO: redesign AsyncJob::status() API to avoid
 436     // id and stop reason reporting duplication.
 437     buf.append(" [state:", 8);
 438     if (requestWritten) buf.append("w", 1); // request sent
 439     if (tunnelEstablished) buf.append("t", 1); // tunnel established
 440     if (!callback) buf.append("x", 1); // caller informed
 441     if (stopReason != nullptr) {
 442         buf.append(" stopped, reason:", 16);
 443         buf.appendf("%s",stopReason);
 444     }
 445     if (connection != nullptr)
 446         buf.appendf(" FD %d", connection->fd);
 447     buf.appendf(" %s%u]", id.prefix(), id.value);
 448     buf.terminate();
 449
 450     return buf.content();
 451 }
 452