src/icp_v2.cc

   1 /*
   2  * DEBUG: section 12    Internet Cache Protocol (ICP)
   3  * AUTHOR: Duane Wessels
   4  *
   5  * SQUID Web Proxy Cache          http://www.squid-cache.org/
   6  * ----------------------------------------------------------
   7  *
   8  *  Squid is the result of efforts by numerous individuals from
   9  *  the Internet community; see the CONTRIBUTORS file for full
  10  *  details.   Many organizations have provided support for Squid's
  11  *  development; see the SPONSORS file for full details.  Squid is
  12  *  Copyrighted (C) 2001 by the Regents of the University of
  13  *  California; see the COPYRIGHT file for full details.  Squid
  14  *  incorporates software developed and/or copyrighted by other
  15  *  sources; see the CREDITS file for full details.
  16  *
  17  *  This program is free software; you can redistribute it and/or modify
  18  *  it under the terms of the GNU General Public License as published by
  19  *  the Free Software Foundation; either version 2 of the License, or
  20  *  (at your option) any later version.
  21  *
  22  *  This program is distributed in the hope that it will be useful,
  23  *  but WITHOUT ANY WARRANTY; without even the implied warranty of
  24  *  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  25  *  GNU General Public License for more details.
  26  *
  27  *  You should have received a copy of the GNU General Public License
  28  *  along with this program; if not, write to the Free Software
  29  *  Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA 02111, USA.
  30  *
  31  */
  32
  33 /**
  34  \defgroup ServerProtocolICPInternal2 ICPv2 Internals
  35  \ingroup ServerProtocolICPAPI
  36  */
  37
  38 #include "squid.h"
  39 #include "Store.h"
  40 #include "comm.h"
  41 #include "comm/Loops.h"
  42 #include "ICP.h"
  43 #include "comm/Connection.h"
  44 #include "HttpRequest.h"
  45 #include "acl/FilledChecklist.h"
  46 #include "acl/Acl.h"
  47 #include "AccessLogEntry.h"
  48 #include "wordlist.h"
  49 #include "SquidTime.h"
  50 #include "SwapDir.h"
  51 #include "icmp/net_db.h"
  52 #include "ip/Address.h"
  53 #include "ip/tools.h"
  54 #include "ipc/StartListening.h"
  55 #include "ipcache.h"
  56 #include "rfc1738.h"
  57
  58 /// dials icpIncomingConnectionOpened call
  59 class IcpListeningStartedDialer: public CallDialer,
  60         public Ipc::StartListeningCb
  61 {
  62 public:
  63     typedef void (*Handler)(int errNo);
  64     IcpListeningStartedDialer(Handler aHandler):
  65             handler(aHandler) {}
  66
  67     virtual void print(std::ostream &os) const { startPrint(os) << ')'; }
  68     virtual bool canDial(AsyncCall &) const { return true; }
  69     virtual void dial(AsyncCall &) { (handler)(errNo); }
  70
  71 public:
  72     Handler handler;
  73 };
  74
  75 static void icpIncomingConnectionOpened(int errNo);
  76
  77 /// \ingroup ServerProtocolICPInternal2
  78 static void icpLogIcp(const Ip::Address &, log_type, int, const char *, int);
  79
  80 /// \ingroup ServerProtocolICPInternal2
  81 static void icpHandleIcpV2(int, Ip::Address &, char *, int);
  82
  83 /// \ingroup ServerProtocolICPInternal2
  84 static void icpCount(void *, int, size_t, int);
  85
  86 /// \ingroup ServerProtocolICPInternal2
  87 static void icpGetOutgoingIpAddress();
  88
  89 /**
  90  \ingroup ServerProtocolICPInternal2
  91  * IcpQueueHead is global so comm_incoming() knows whether or not
  92  * to call icpUdpSendQueue.
  93  */
  94 static icpUdpData *IcpQueueHead = NULL;
  95 /// \ingroup ServerProtocolICPInternal2
  96 static icpUdpData *IcpQueueTail = NULL;
  97
  98 /// \ingroup ServerProtocolICPInternal2
  99 Comm::ConnectionPointer icpIncomingConn = NULL;
 100 /// \ingroup ServerProtocolICPInternal2
 101 Comm::ConnectionPointer icpOutgoingConn = NULL;
 102
 103 /** \ingroup ServerProtocolICPInternal2
 104  * ICP v2 uses the outgoing address as host ID.
 105  * NP: this *may* be identical to icpOutgoingConn->local
 106  * but when IN/OUT sockets are shared we can't guarantee that
 107  * so a separate variable is used for now.
 108  *
 109  * We have one for private use (sent only by this local cache)
 110  * and one for public use (for external caches to contact us)
 111  */
 112 Ip::Address theIcpPrivateHostID;
 113
 114 /// \see theIcpPrivateHostID
 115 Ip::Address theIcpPublicHostID;
 116
 117
 118 /* icp_common_t */
 119 _icp_common_t::_icp_common_t() : opcode(ICP_INVALID), version(0), length(0), reqnum(0), flags(0), pad(0), shostid(0)
 120 {}
 121
 122 _icp_common_t::_icp_common_t(char *buf, unsigned int len)
 123 {
 124     if (len < sizeof(_icp_common_t)) {
 125         /* mark as invalid */
 126         length = len + 1;
 127         return;
 128     }
 129
 130     memcpy(this, buf, sizeof(icp_common_t));
 131     /*
 132      * Convert network order sensitive fields
 133      */
 134     length = ntohs(length);
 135     reqnum = ntohl(reqnum);
 136     flags = ntohl(flags);
 137     pad = ntohl(pad);
 138 }
 139
 140 icp_opcode
 141 _icp_common_t::getOpCode() const
 142 {
 143     if (opcode > (char)ICP_END)
 144         return ICP_INVALID;
 145
 146     return (icp_opcode)opcode;
 147 }
 148
 149 /* ICPState */
 150
 151 ICPState::ICPState(icp_common_t &aHeader, HttpRequest *aRequest):
 152         header(aHeader),
 153         request(HTTPMSGLOCK(aRequest)),
 154         fd(-1),
 155         url(NULL)
 156 {}
 157
 158 ICPState::~ICPState()
 159 {
 160     safe_free(url);
 161     HTTPMSGUNLOCK(request);
 162 }
 163
 164
 165 /* End ICPState */
 166
 167 /* ICP2State */
 168
 169 /// \ingroup ServerProtocolICPInternal2
 170 class ICP2State : public ICPState, public StoreClient
 171 {
 172
 173 public:
 174     ICP2State(icp_common_t & aHeader, HttpRequest *aRequest):
 175             ICPState(aHeader, aRequest),rtt(0),src_rtt(0),flags(0) {}
 176
 177     ~ICP2State();
 178     void created(StoreEntry * newEntry);
 179
 180     int rtt;
 181     int src_rtt;
 182     uint32_t flags;
 183 };
 184
 185 ICP2State::~ICP2State()
 186 {}
 187
 188 void
 189 ICP2State::created(StoreEntry *newEntry)
 190 {
 191     StoreEntry *entry = newEntry->isNull () ? NULL : newEntry;
 192     debugs(12, 5, "icpHandleIcpV2: OPCODE " << icp_opcode_str[header.opcode]);
 193     icp_opcode codeToSend;
 194
 195     if (icpCheckUdpHit(entry, request)) {
 196         codeToSend = ICP_HIT;
 197     } else {
 198 #if USE_ICMP
 199         if (Config.onoff.test_reachability && rtt == 0) {
 200             if ((rtt = netdbHostRtt(request->GetHost())) == 0)
 201                 netdbPingSite(request->GetHost());
 202         }
 203 #endif /* USE_ICMP */
 204
 205         if (icpGetCommonOpcode() != ICP_ERR)
 206             codeToSend = icpGetCommonOpcode();
 207         else if (Config.onoff.test_reachability && rtt == 0)
 208             codeToSend = ICP_MISS_NOFETCH;
 209         else
 210             codeToSend = ICP_MISS;
 211     }
 212
 213     icpCreateAndSend(codeToSend, flags, url, header.reqnum, src_rtt, fd, from);
 214     delete this;
 215 }
 216
 217 /* End ICP2State */
 218
 219 /// \ingroup ServerProtocolICPInternal2
 220 static void
 221 icpLogIcp(const Ip::Address &caddr, log_type logcode, int len, const char *url, int delay)
 222 {
 223     AccessLogEntry al;
 224
 225     if (LOG_TAG_NONE == logcode)
 226         return;
 227
 228     if (LOG_ICP_QUERY == logcode)
 229         return;
 230
 231     clientdbUpdate(caddr, logcode, AnyP::PROTO_ICP, len);
 232
 233     if (!Config.onoff.log_udp)
 234         return;
 235
 236     al.icp.opcode = ICP_QUERY;
 237
 238     al.url = url;
 239
 240     al.cache.caddr = caddr;
 241
 242     al.cache.replySize = len;
 243
 244     al.cache.code = logcode;
 245
 246     al.cache.msec = delay;
 247
 248     accessLogLog(&al, NULL);
 249 }
 250
 251 /// \ingroup ServerProtocolICPInternal2
 252 void
 253 icpUdpSendQueue(int fd, void *unused)
 254 {
 255     icpUdpData *q;
 256
 257     while ((q = IcpQueueHead) != NULL) {
 258         int delay = tvSubUsec(q->queue_time, current_time);
 259         /* increment delay to prevent looping */
 260         const int x = icpUdpSend(fd, q->address, (icp_common_t *) q->msg, q->logcode, ++delay);
 261         IcpQueueHead = q->next;
 262         xfree(q);
 263
 264         if (x < 0)
 265             break;
 266     }
 267 }
 268
 269 _icp_common_t *
 270 _icp_common_t::createMessage(
 271     icp_opcode opcode,
 272     int flags,
 273     const char *url,
 274     int reqnum,
 275     int pad)
 276 {
 277     char *buf = NULL;
 278     icp_common_t *headerp = NULL;
 279     char *urloffset = NULL;
 280     int buf_len;
 281     buf_len = sizeof(icp_common_t) + strlen(url) + 1;
 282
 283     if (opcode == ICP_QUERY)
 284         buf_len += sizeof(uint32_t);
 285
 286     buf = (char *) xcalloc(buf_len, 1);
 287
 288     headerp = (icp_common_t *) (void *) buf;
 289
 290     headerp->opcode = (char) opcode;
 291
 292     headerp->version = ICP_VERSION_CURRENT;
 293
 294     headerp->length = (uint16_t) htons(buf_len);
 295
 296     headerp->reqnum = htonl(reqnum);
 297
 298     headerp->flags = htonl(flags);
 299
 300     headerp->pad = htonl(pad);
 301
 302     theIcpPrivateHostID.GetInAddr( *((struct in_addr*)&headerp->shostid) );
 303
 304     urloffset = buf + sizeof(icp_common_t);
 305
 306     if (opcode == ICP_QUERY)
 307         urloffset += sizeof(uint32_t);
 308
 309     memcpy(urloffset, url, strlen(url));
 310
 311     return (icp_common_t *)buf;
 312 }
 313
 314 int
 315 icpUdpSend(int fd,
 316            const Ip::Address &to,
 317            icp_common_t * msg,
 318            log_type logcode,
 319            int delay)
 320 {
 321     icpUdpData *queue;
 322     int x;
 323     int len;
 324     len = (int) ntohs(msg->length);
 325     debugs(12, 5, "icpUdpSend: FD " << fd << " sending " <<
 326            icp_opcode_str[msg->opcode] << ", " << len << " bytes to " << to);
 327
 328     x = comm_udp_sendto(fd, to, msg, len);
 329
 330     if (x >= 0) {
 331         /* successfully written */
 332         icpLogIcp(to, logcode, len, (char *) (msg + 1), delay);
 333         icpCount(msg, SENT, (size_t) len, delay);
 334         safe_free(msg);
 335     } else if (0 == delay) {
 336         /* send failed, but queue it */
 337         queue = (icpUdpData *) xcalloc(1, sizeof(icpUdpData));
 338         queue->address = to;
 339         queue->msg = msg;
 340         queue->len = (int) ntohs(msg->length);
 341         queue->queue_time = current_time;
 342         queue->logcode = logcode;
 343
 344         if (IcpQueueHead == NULL) {
 345             IcpQueueHead = queue;
 346             IcpQueueTail = queue;
 347         } else if (IcpQueueTail == IcpQueueHead) {
 348             IcpQueueTail = queue;
 349             IcpQueueHead->next = queue;
 350         } else {
 351             IcpQueueTail->next = queue;
 352             IcpQueueTail = queue;
 353         }
 354
 355         Comm::SetSelect(fd, COMM_SELECT_WRITE, icpUdpSendQueue, NULL, 0);
 356         statCounter.icp.replies_queued++;
 357     } else {
 358         /* don't queue it */
 359         statCounter.icp.replies_dropped++;
 360     }
 361
 362     return x;
 363 }
 364
 365 int
 366 icpCheckUdpHit(StoreEntry * e, HttpRequest * request)
 367 {
 368     if (e == NULL)
 369         return 0;
 370
 371     if (!e->validToSend())
 372         return 0;
 373
 374     if (Config.onoff.icp_hit_stale)
 375         return 1;
 376
 377     if (refreshCheckICP(e, request))
 378         return 0;
 379
 380     return 1;
 381 }
 382
 383 /**
 384  * This routine selects an ICP opcode for ICP misses.
 385  *
 386  \retval ICP_ERR            no opcode selected here
 387  \retval ICP_MISS_NOFETCH   store is rebuilding, no fetch is possible yet
 388  */
 389 icp_opcode
 390 icpGetCommonOpcode()
 391 {
 392     /* if store is rebuilding, return a UDP_MISS_NOFETCH */
 393
 394     if ((StoreController::store_dirs_rebuilding && opt_reload_hit_only) ||
 395             hit_only_mode_until > squid_curtime) {
 396         return ICP_MISS_NOFETCH;
 397     }
 398
 399     return ICP_ERR;
 400 }
 401
 402 log_type
 403 icpLogFromICPCode(icp_opcode opcode)
 404 {
 405     if (opcode == ICP_ERR)
 406         return LOG_UDP_INVALID;
 407
 408     if (opcode == ICP_DENIED)
 409         return LOG_UDP_DENIED;
 410
 411     if (opcode == ICP_HIT)
 412         return LOG_UDP_HIT;
 413
 414     if (opcode == ICP_MISS)
 415         return LOG_UDP_MISS;
 416
 417     if (opcode == ICP_MISS_NOFETCH)
 418         return LOG_UDP_MISS_NOFETCH;
 419
 420     fatal("expected ICP opcode\n");
 421
 422     return LOG_UDP_INVALID;
 423 }
 424
 425 void
 426 icpCreateAndSend(icp_opcode opcode, int flags, char const *url, int reqnum, int pad, int fd, const Ip::Address &from)
 427 {
 428     icp_common_t *reply = _icp_common_t::createMessage(opcode, flags, url, reqnum, pad);
 429     icpUdpSend(fd, from, reply, icpLogFromICPCode(opcode), 0);
 430 }
 431
 432 void
 433 icpDenyAccess(Ip::Address &from, char *url, int reqnum, int fd)
 434 {
 435     debugs(12, 2, "icpDenyAccess: Access Denied for " << from << " by " << AclMatchedName << ".");
 436
 437     if (clientdbCutoffDenied(from)) {
 438         /*
 439          * count this DENIED query in the clientdb, even though
 440          * we're not sending an ICP reply...
 441          */
 442         clientdbUpdate(from, LOG_UDP_DENIED, AnyP::PROTO_ICP, 0);
 443     } else {
 444         icpCreateAndSend(ICP_DENIED, 0, url, reqnum, 0, fd, from);
 445     }
 446 }
 447
 448 bool
 449 icpAccessAllowed(Ip::Address &from, HttpRequest * icp_request)
 450 {
 451     /* absent an explicit allow, we deny all */
 452     if (!Config.accessList.icp)
 453         return true;
 454
 455     ACLFilledChecklist checklist(Config.accessList.icp, icp_request, NULL);
 456     checklist.src_addr = from;
 457     checklist.my_addr.SetNoAddr();
 458     return (checklist.fastCheck() == ACCESS_ALLOWED);
 459 }
 460
 461 char const *
 462 icpGetUrlToSend(char *url)
 463 {
 464     if (strpbrk(url, w_space))
 465         return rfc1738_escape(url);
 466     else
 467         return url;
 468 }
 469
 470 HttpRequest *
 471 icpGetRequest(char *url, int reqnum, int fd, Ip::Address &from)
 472 {
 473     if (strpbrk(url, w_space)) {
 474         url = rfc1738_escape(url);
 475         icpCreateAndSend(ICP_ERR, 0, rfc1738_escape(url), reqnum, 0, fd, from);
 476         return NULL;
 477     }
 478
 479     HttpRequest *result;
 480
 481     if ((result = HttpRequest::CreateFromUrl(url)) == NULL)
 482         icpCreateAndSend(ICP_ERR, 0, url, reqnum, 0, fd, from);
 483
 484     return result;
 485
 486 }
 487
 488 static void
 489 doV2Query(int fd, Ip::Address &from, char *buf, icp_common_t header)
 490 {
 491     int rtt = 0;
 492     int src_rtt = 0;
 493     uint32_t flags = 0;
 494     /* We have a valid packet */
 495     char *url = buf + sizeof(icp_common_t) + sizeof(uint32_t);
 496     HttpRequest *icp_request = icpGetRequest(url, header.reqnum, fd, from);
 497
 498     if (!icp_request)
 499         return;
 500
 501     HTTPMSGLOCK(icp_request);
 502
 503     if (!icpAccessAllowed(from, icp_request)) {
 504         icpDenyAccess(from, url, header.reqnum, fd);
 505         HTTPMSGUNLOCK(icp_request);
 506         return;
 507     }
 508 #if USE_ICMP
 509     if (header.flags & ICP_FLAG_SRC_RTT) {
 510         rtt = netdbHostRtt(icp_request->GetHost());
 511         int hops = netdbHostHops(icp_request->GetHost());
 512         src_rtt = ((hops & 0xFFFF) << 16) | (rtt & 0xFFFF);
 513
 514         if (rtt)
 515             flags |= ICP_FLAG_SRC_RTT;
 516     }
 517 #endif /* USE_ICMP */
 518
 519     /* The peer is allowed to use this cache */
 520     ICP2State *state = new ICP2State (header, icp_request);
 521
 522     state->fd = fd;
 523
 524     state->from = from;
 525
 526     state->url = xstrdup (url);
 527
 528     state->flags = flags;
 529
 530     state->rtt = rtt;
 531
 532     state->src_rtt = src_rtt;
 533
 534     StoreEntry::getPublic (state, url, METHOD_GET);
 535
 536     HTTPMSGUNLOCK(icp_request);
 537 }
 538
 539 void
 540 _icp_common_t::handleReply(char *buf, Ip::Address &from)
 541 {
 542     if (neighbors_do_private_keys && reqnum == 0) {
 543         debugs(12, 0, "icpHandleIcpV2: Neighbor " << from << " returned reqnum = 0");
 544         debugs(12, 0, "icpHandleIcpV2: Disabling use of private keys");
 545         neighbors_do_private_keys = 0;
 546     }
 547
 548     char *url = buf + sizeof(icp_common_t);
 549     debugs(12, 3, "icpHandleIcpV2: " << icp_opcode_str[opcode] << " from " << from << " for '" << url << "'");
 550
 551     const cache_key *key = icpGetCacheKey(url, (int) reqnum);
 552     /* call neighborsUdpAck even if ping_status != PING_WAITING */
 553     neighborsUdpAck(key, this, from);
 554 }
 555
 556 static void
 557 icpHandleIcpV2(int fd, Ip::Address &from, char *buf, int len)
 558 {
 559     if (len <= 0) {
 560         debugs(12, 3, "icpHandleIcpV2: ICP message is too small");
 561         return;
 562     }
 563
 564     icp_common_t header(buf, len);
 565     /*
 566      * Length field should match the number of bytes read
 567      */
 568
 569     if (len != header.length) {
 570         debugs(12, 3, "icpHandleIcpV2: ICP message is too small");
 571         return;
 572     }
 573
 574     switch (header.opcode) {
 575
 576     case ICP_QUERY:
 577         /* We have a valid packet */
 578         doV2Query(fd, from, buf, header);
 579         break;
 580
 581     case ICP_HIT:
 582
 583     case ICP_DECHO:
 584
 585     case ICP_MISS:
 586
 587     case ICP_DENIED:
 588
 589     case ICP_MISS_NOFETCH:
 590         header.handleReply(buf, from);
 591         break;
 592
 593     case ICP_INVALID:
 594
 595     case ICP_ERR:
 596         break;
 597
 598     default:
 599         debugs(12, 0, "icpHandleIcpV2: UNKNOWN OPCODE: " << header.opcode << " from " << from);
 600
 601         break;
 602     }
 603 }
 604
 605 #ifdef ICP_PKT_DUMP
 606 static void
 607 icpPktDump(icp_common_t * pkt)
 608 {
 609     Ip::Address a;
 610
 611     debugs(12, 9, "opcode:     " << std::setw(3) << pkt->opcode  << " " << icp_opcode_str[pkt->opcode]);
 612     debugs(12, 9, "version: "<< std::left << std::setw(8) << pkt->version);
 613     debugs(12, 9, "length:  "<< std::left << std::setw(8) << ntohs(pkt->length));
 614     debugs(12, 9, "reqnum:  "<< std::left << std::setw(8) << ntohl(pkt->reqnum));
 615     debugs(12, 9, "flags:   "<< std::left << std::hex << std::setw(8) << ntohl(pkt->flags));
 616     a = (struct in_addr)pkt->shostid;
 617     debugs(12, 9, "shostid: " << a );
 618     debugs(12, 9, "payload: " << (char *) pkt + sizeof(icp_common_t));
 619 }
 620
 621 #endif
 622
 623 void
 624 icpHandleUdp(int sock, void *data)
 625 {
 626     int *N = &incoming_sockets_accepted;
 627
 628     Ip::Address from;
 629     LOCAL_ARRAY(char, buf, SQUID_UDP_SO_RCVBUF);
 630     int len;
 631     int icp_version;
 632     int max = INCOMING_ICP_MAX;
 633     Comm::SetSelect(sock, COMM_SELECT_READ, icpHandleUdp, NULL, 0);
 634
 635     while (max--) {
 636         len = comm_udp_recvfrom(sock,
 637                                 buf,
 638                                 SQUID_UDP_SO_RCVBUF - 1,
 639                                 0,
 640                                 from);
 641
 642         if (len == 0)
 643             break;
 644
 645         if (len < 0) {
 646             if (ignoreErrno(errno))
 647                 break;
 648
 649 #if _SQUID_LINUX_
 650             /* Some Linux systems seem to set the FD for reading and then
 651              * return ECONNREFUSED when sendto() fails and generates an ICMP
 652              * port unreachable message. */
 653             /* or maybe an EHOSTUNREACH "No route to host" message */
 654             if (errno != ECONNREFUSED && errno != EHOSTUNREACH)
 655 #endif
 656
 657                 debugs(50, 1, "icpHandleUdp: FD " << sock << " recvfrom: " << xstrerror());
 658
 659             break;
 660         }
 661
 662         (*N)++;
 663         icpCount(buf, RECV, (size_t) len, 0);
 664         buf[len] = '\0';
 665         debugs(12, 4, "icpHandleUdp: FD " << sock << ": received " <<
 666                (unsigned long int)len << " bytes from " << from);
 667
 668 #ifdef ICP_PACKET_DUMP
 669
 670         icpPktDump(buf);
 671 #endif
 672
 673         if ((size_t) len < sizeof(icp_common_t)) {
 674             debugs(12, 4, "icpHandleUdp: Ignoring too-small UDP packet");
 675             break;
 676         }
 677
 678         icp_version = (int) buf[1];     /* cheat! */
 679
 680         if (icpOutgoingConn->local == from)
 681             // ignore ICP packets which loop back (multicast usually)
 682             debugs(12, 4, "icpHandleUdp: Ignoring UDP packet sent by myself");
 683         else if (icp_version == ICP_VERSION_2)
 684             icpHandleIcpV2(sock, from, buf, len);
 685         else if (icp_version == ICP_VERSION_3)
 686             icpHandleIcpV3(sock, from, buf, len);
 687         else
 688             debugs(12, 1, "WARNING: Unused ICP version " << icp_version <<
 689                    " received from " << from);
 690     }
 691 }
 692
 693 void
 694 icpConnectionsOpen(void)
 695 {
 696     uint16_t port;
 697
 698     if ((port = Config.Port.icp) <= 0)
 699         return;
 700
 701     icpIncomingConn = new Comm::Connection;
 702     icpIncomingConn->local = Config.Addrs.udp_incoming;
 703     icpIncomingConn->local.SetPort(port);
 704
 705     if (!Ip::EnableIpv6 && !icpIncomingConn->local.SetIPv4()) {
 706         debugs(12, DBG_CRITICAL, "ERROR: IPv6 is disabled. " << icpIncomingConn->local << " is not an IPv4 address.");
 707         fatal("ICP port cannot be opened.");
 708     }
 709     /* split-stack for now requires default IPv4-only ICP */
 710     if (Ip::EnableIpv6&IPV6_SPECIAL_SPLITSTACK && icpIncomingConn->local.IsAnyAddr()) {
 711         icpIncomingConn->local.SetIPv4();
 712     }
 713
 714     AsyncCall::Pointer call = asyncCall(12, 2,
 715                                         "icpIncomingConnectionOpened",
 716                                         IcpListeningStartedDialer(&icpIncomingConnectionOpened));
 717
 718     Ipc::StartListening(SOCK_DGRAM,
 719                         IPPROTO_UDP,
 720                         icpIncomingConn,
 721                         Ipc::fdnInIcpSocket, call);
 722
 723     if ( !Config.Addrs.udp_outgoing.IsNoAddr() ) {
 724         icpOutgoingConn = new Comm::Connection;
 725         icpOutgoingConn->local = Config.Addrs.udp_outgoing;
 726         icpOutgoingConn->local.SetPort(port);
 727
 728         if (!Ip::EnableIpv6 && !icpOutgoingConn->local.SetIPv4()) {
 729             debugs(49, DBG_CRITICAL, "ERROR: IPv6 is disabled. " << icpOutgoingConn->local << " is not an IPv4 address.");
 730             fatal("ICP port cannot be opened.");
 731         }
 732         /* split-stack for now requires default IPv4-only ICP */
 733         if (Ip::EnableIpv6&IPV6_SPECIAL_SPLITSTACK && icpOutgoingConn->local.IsAnyAddr()) {
 734             icpOutgoingConn->local.SetIPv4();
 735         }
 736
 737         enter_suid();
 738         comm_open_listener(SOCK_DGRAM, IPPROTO_UDP, icpOutgoingConn, "Outgoing ICP Port");
 739         leave_suid();
 740
 741         if (!Comm::IsConnOpen(icpOutgoingConn))
 742             fatal("Cannot open Outgoing ICP Port");
 743
 744         debugs(12, DBG_CRITICAL, "Sending ICP messages from " << icpOutgoingConn->local);
 745
 746         Comm::SetSelect(icpOutgoingConn->fd, COMM_SELECT_READ, icpHandleUdp, NULL, 0);
 747         fd_note(icpOutgoingConn->fd, "Outgoing ICP socket");
 748         icpGetOutgoingIpAddress();
 749     }
 750 }
 751
 752 // Ensure that we have the IP address(es) to use for Host ID.
 753 // The outgoing address is used as 'private' host ID used only on packets we send
 754 static void
 755 icpGetOutgoingIpAddress()
 756 {
 757     struct addrinfo *xai = NULL;
 758     theIcpPrivateHostID.SetEmpty();
 759     theIcpPrivateHostID.InitAddrInfo(xai);
 760     if (getsockname(icpOutgoingConn->fd, xai->ai_addr, &xai->ai_addrlen) < 0)
 761         debugs(50, DBG_IMPORTANT, "ERROR: Unable to identify ICP host ID to use for " << icpOutgoingConn
 762                << ": getsockname: " << xstrerror());
 763     else
 764         theIcpPrivateHostID = *xai;
 765     theIcpPrivateHostID.FreeAddrInfo(xai);
 766 }
 767
 768 static void
 769 icpIncomingConnectionOpened(int errNo)
 770 {
 771     if (!Comm::IsConnOpen(icpIncomingConn))
 772         fatal("Cannot open ICP Port");
 773
 774     Comm::SetSelect(icpIncomingConn->fd, COMM_SELECT_READ, icpHandleUdp, NULL, 0);
 775
 776     for (const wordlist *s = Config.mcast_group_list; s; s = s->next)
 777         ipcache_nbgethostbyname(s->key, mcastJoinGroups, NULL); // XXX: pass the icpIncomingConn for mcastJoinGroups usage.
 778
 779     debugs(12, DBG_IMPORTANT, "Accepting ICP messages on " << icpIncomingConn->local);
 780
 781     fd_note(icpIncomingConn->fd, "Incoming ICP port");
 782
 783     if (Config.Addrs.udp_outgoing.IsNoAddr()) {
 784         icpOutgoingConn = icpIncomingConn;
 785         debugs(12, DBG_IMPORTANT, "Sending ICP messages from " << icpOutgoingConn->local);
 786         icpGetOutgoingIpAddress();
 787     }
 788
 789     // Ensure that we have the IP address(es) to use for Host ID.
 790     // The listening address is used as 'public' host ID which can be used to contact us
 791     struct addrinfo *xai = NULL;
 792     theIcpPublicHostID.InitAddrInfo(xai); // reset xai
 793     if (getsockname(icpIncomingConn->fd, xai->ai_addr, &xai->ai_addrlen) < 0)
 794         debugs(50, DBG_IMPORTANT, "ERROR: Unable to identify ICP host ID to use for " << icpIncomingConn
 795                << ": getsockname: " << xstrerror());
 796     else
 797         theIcpPublicHostID = *xai;
 798     theIcpPublicHostID.FreeAddrInfo(xai);
 799 }
 800
 801 /**
 802  * icpConnectionShutdown only closes the 'in' socket if it is
 803  * different than the 'out' socket.
 804  */
 805 void
 806 icpConnectionShutdown(void)
 807 {
 808     if (!Comm::IsConnOpen(icpIncomingConn))
 809         return;
 810
 811     debugs(12, DBG_IMPORTANT, "Stop receiving ICP on " << icpIncomingConn->local);
 812
 813     /** Release the 'in' socket for lazy closure.
 814      * in and out sockets may be sharing one same FD.
 815      * This prevents this function from executing repeatedly.
 816      */
 817     icpIncomingConn = NULL;
 818
 819     /**
 820      * Normally we only write to the outgoing ICP socket, but
 821      * we also have a read handler there to catch messages sent
 822      * to that specific interface.  During shutdown, we must
 823      * disable reading on the outgoing socket.
 824      */
 825     assert(Comm::IsConnOpen(icpOutgoingConn));
 826
 827     Comm::SetSelect(icpOutgoingConn->fd, COMM_SELECT_READ, NULL, NULL, 0);
 828 }
 829
 830 void
 831 icpConnectionClose(void)
 832 {
 833     icpConnectionShutdown();
 834
 835     if (icpOutgoingConn != NULL) {
 836         debugs(12, DBG_IMPORTANT, "Stop sending ICP from " << icpOutgoingConn->local);
 837         icpOutgoingConn = NULL;
 838     }
 839 }
 840
 841 static void
 842 icpCount(void *buf, int which, size_t len, int delay)
 843 {
 844     icp_common_t *icp = (icp_common_t *) buf;
 845
 846     if (len < sizeof(*icp))
 847         return;
 848
 849     if (SENT == which) {
 850         statCounter.icp.pkts_sent++;
 851         kb_incr(&statCounter.icp.kbytes_sent, len);
 852
 853         if (ICP_QUERY == icp->opcode) {
 854             statCounter.icp.queries_sent++;
 855             kb_incr(&statCounter.icp.q_kbytes_sent, len);
 856         } else {
 857             statCounter.icp.replies_sent++;
 858             kb_incr(&statCounter.icp.r_kbytes_sent, len);
 859             /* this is the sent-reply service time */
 860             statHistCount(&statCounter.icp.reply_svc_time, delay);
 861         }
 862
 863         if (ICP_HIT == icp->opcode)
 864             statCounter.icp.hits_sent++;
 865     } else if (RECV == which) {
 866         statCounter.icp.pkts_recv++;
 867         kb_incr(&statCounter.icp.kbytes_recv, len);
 868
 869         if (ICP_QUERY == icp->opcode) {
 870             statCounter.icp.queries_recv++;
 871             kb_incr(&statCounter.icp.q_kbytes_recv, len);
 872         } else {
 873             statCounter.icp.replies_recv++;
 874             kb_incr(&statCounter.icp.r_kbytes_recv, len);
 875             /* statCounter.icp.query_svc_time set in clientUpdateCounters */
 876         }
 877
 878         if (ICP_HIT == icp->opcode)
 879             statCounter.icp.hits_recv++;
 880     }
 881 }
 882
 883 #define N_QUERIED_KEYS 8192
 884 #define N_QUERIED_KEYS_MASK 8191
 885 static cache_key queried_keys[N_QUERIED_KEYS][SQUID_MD5_DIGEST_LENGTH];
 886
 887 int
 888 icpSetCacheKey(const cache_key * key)
 889 {
 890     static int reqnum = 0;
 891
 892     if (++reqnum < 0)
 893         reqnum = 1;
 894
 895     storeKeyCopy(queried_keys[reqnum & N_QUERIED_KEYS_MASK], key);
 896
 897     return reqnum;
 898 }
 899
 900 const cache_key *
 901 icpGetCacheKey(const char *url, int reqnum)
 902 {
 903     if (neighbors_do_private_keys && reqnum)
 904         return queried_keys[reqnum & N_QUERIED_KEYS_MASK];
 905
 906     return storeKeyPublic(url, METHOD_GET);
 907 }