diff --git a/src/comsock.c b/src/comsock.c index 1873be621..1d63e7314 100644 --- a/src/comsock.c +++ b/src/comsock.c @@ -1,4 +1,4 @@ -// Copyright 2015-2021 The NATS Authors +// Copyright 2015-2025 The NATS Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at @@ -134,6 +134,18 @@ natsSock_ShuffleIPs(natsSockCtx *ctx, struct addrinfo **tmp, int tmpSize, struct #define MAX_HOST_NAME (256) +void resetDeadline(natsSockCtx* ctx, int64_t start, int64_t totalTimeout) +{ + // If there was a deadline, reset the deadline with whatever is left. + if (totalTimeout > 0) + { + int64_t used = nats_Now() - start; + int64_t left = totalTimeout - used; + + natsDeadline_Init(&(ctx->writeDeadline), (left > 0 ? left : 0)); + } +} + natsStatus natsSock_ConnectTcp(natsSockCtx *ctx, const char *phost, int port) { @@ -152,6 +164,7 @@ natsSock_ConnectTcp(natsSockCtx *ctx, const char *phost, int port) int64_t totalTimeout = 0; int64_t timeoutPerIP = 0; struct addrinfo *tmpStorage[64]; + bool hasProxyConnectCb = ctx->proxyConnectCb != NULL; if (phost == NULL) return nats_setError(NATS_ADDRESS_MISSING, "%s", "No host specified"); @@ -170,44 +183,55 @@ natsSock_ConnectTcp(natsSockCtx *ctx, const char *phost, int port) snprintf(sport, sizeof(sport), "%d", port); - if ((ctx->orderIP == 46) || (ctx->orderIP == 64)) - max = 2; - start = nats_Now(); - for (i=0; iproxyConnectCb(host, port, &ctx->fd); - memset(&hints,0,sizeof(hints)); - hints.ai_socktype = SOCK_STREAM; + resetDeadline(ctx, start, totalTimeout); - switch (ctx->orderIP) - { - case 4: hints.ai_family = AF_INET; break; - case 6: hints.ai_family = AF_INET6; break; - case 46: hints.ai_family = (i == 0 ? AF_INET : AF_INET6); break; - case 64: hints.ai_family = (i == 0 ? AF_INET6 : AF_INET); break; - default: hints.ai_family = AF_UNSPEC; - } + return NATS_UPDATE_ERR_STACK(s); + } - if ((res = getaddrinfo(host, sport, &hints, &servinfo)) != 0) - { - s = nats_setError(NATS_SYS_ERROR, "getaddrinfo error: %s", - gai_strerror(res)); - continue; + if ((ctx->orderIP == 46) || (ctx->orderIP == 64)) + max = 2; + + for (i = 0; i < max; i++) + { + struct addrinfo hints; + struct addrinfo* servinfo = NULL; + int count = 0; + struct addrinfo* p; + + memset(&hints, 0, sizeof(hints)); + hints.ai_socktype = SOCK_STREAM; + + switch (ctx->orderIP) + { + case 4: hints.ai_family = AF_INET; break; + case 6: hints.ai_family = AF_INET6; break; + case 46: hints.ai_family = (i == 0 ? AF_INET : AF_INET6); break; + case 64: hints.ai_family = (i == 0 ? AF_INET6 : AF_INET); break; + default: hints.ai_family = AF_UNSPEC; } - servInfos[numServInfo] = servinfo; - for (p = servinfo; (p != NULL); p = p->ai_next) - { - count++; - numIPs++; + + if ((res = getaddrinfo(host, sport, &hints, &servinfo)) != 0) + { + s = nats_setError(NATS_SYS_ERROR, "getaddrinfo error: %s", gai_strerror(res)); + continue; } - natsSock_ShuffleIPs(ctx, tmpStorage, sizeof(tmpStorage), &(servInfos[numServInfo]), count); - numServInfo++; + + servInfos[numServInfo] = servinfo; + for (p = servinfo; (p != NULL); p = p->ai_next) + { + count++; + numIPs++; + } + natsSock_ShuffleIPs(ctx, tmpStorage, sizeof(tmpStorage), &(servInfos[numServInfo]), count); + numServInfo++; } // If we got a getaddrinfo() and there is no servInfos to try to connect to // bail out now. @@ -238,7 +262,7 @@ natsSock_ConnectTcp(natsSockCtx *ctx, const char *phost, int port) for (i=0; iai_next) { ctx->fd = socket(p->ai_family, p->ai_socktype, p->ai_protocol); @@ -281,7 +305,7 @@ natsSock_ConnectTcp(natsSockCtx *ctx, const char *phost, int port) s = nats_setDefaultError(NATS_NO_SERVER); } } - + if (s == NATS_OK) { s = natsSock_SetCommonTcpOptions(ctx->fd); @@ -289,7 +313,7 @@ natsSock_ConnectTcp(natsSockCtx *ctx, const char *phost, int port) if (s == NATS_OK) break; } - + _closeFd(ctx->fd); ctx->fd = NATS_SOCK_INVALID; } @@ -301,17 +325,11 @@ natsSock_ConnectTcp(natsSockCtx *ctx, const char *phost, int port) break; } } + for (i=0; i 0) - { - int64_t used = nats_Now() - start; - int64_t left = totalTimeout - used; - - natsDeadline_Init(&(ctx->writeDeadline), (left > 0 ? left : 0)); - } + resetDeadline(ctx, start, totalTimeout); return NATS_UPDATE_ERR_STACK(s); } diff --git a/src/conn.c b/src/conn.c index 659d9d95e..065fe81bd 100644 --- a/src/conn.c +++ b/src/conn.c @@ -1,4 +1,4 @@ -// Copyright 2015-2024 The NATS Authors +// Copyright 2015-2025 The NATS Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at @@ -402,6 +402,9 @@ _createConn(natsConnection *nc) // Set ctx.noRandomize based on public NoRandomize option. nc->sockCtx.noRandomize = nc->opts->noRandomize; + + // Set the proxy connect callback + nc->sockCtx.proxyConnectCb = nc->opts->proxyConnectCb; s = natsSock_ConnectTcp(&(nc->sockCtx), nc->cur->url->host, nc->cur->url->port); if (s == NATS_OK) diff --git a/src/nats.h b/src/nats.h index fd68fc62c..8dfb60320 100644 --- a/src/nats.h +++ b/src/nats.h @@ -1672,6 +1672,15 @@ typedef void (*natsMsgHandler)( typedef void (*natsConnectionHandler)( natsConnection *nc, void *closure); +/** \brief Callback used to handle connections via proxy. + * + * This callback is used to handle connections via proxy. + * It creates a socket, use it for proxy verification, and return it in `fd` + * to be used for the bus connection. + */ +typedef natsStatus (*natsProxyConnHandler)( + char* host, int port, natsSock* fd); + /** \brief Callback used to notify the user of errors encountered while processing * inbound messages. * @@ -2951,6 +2960,18 @@ natsOptions_SetMaxPendingMsgs(natsOptions *opts, int maxPending); NATS_EXTERN natsStatus natsOptions_SetMaxPendingBytes(natsOptions* opts, int64_t maxPending); +/** \brief Sets the proxy connection handler. + * + * Specifies the callback to invoke for proxy connection returning the socket to use. + * + * @see natsProxyConnHandler + * + * @param opts the pointer to the #natsOptions object. + * @param proxyConnHandler the proxy connection handler callback. + */ +NATS_EXTERN natsStatus +natsOptions_SetProxyConnHandler(natsOptions* opts, natsProxyConnHandler proxyConnHandler); + /** \brief Sets the error handler for asynchronous events. * * Specifies the callback to invoke when an asynchronous error diff --git a/src/natsp.h b/src/natsp.h index 7ec1b912b..160b4628a 100644 --- a/src/natsp.h +++ b/src/natsp.h @@ -277,6 +277,8 @@ struct __natsOptions natsConnectionHandler microClosedCb; natsErrHandler microAsyncErrCb; + natsProxyConnHandler proxyConnectCb; + int64_t pingInterval; int maxPingsOut; int maxPendingMsgs; @@ -653,22 +655,24 @@ typedef struct __natsPongList typedef struct __natsSockCtx { - natsSock fd; - bool fdActive; + natsSock fd; + bool fdActive; - natsDeadline readDeadline; - natsDeadline writeDeadline; + natsDeadline readDeadline; + natsDeadline writeDeadline; - SSL *ssl; + SSL *ssl; // This is true when we are using an external event loop (such as libuv). - bool useEventLoop; + bool useEventLoop; - int orderIP; // possible values: 0,4,6,46,64 + int orderIP; // possible values: 0,4,6,46,64 // By default, the list of IPs returned by the hostname resolution will // be shuffled. This option, if `true`, will disable the shuffling. - bool noRandomize; + bool noRandomize; + + natsProxyConnHandler proxyConnectCb; } natsSockCtx; diff --git a/src/opts.c b/src/opts.c index 48c75382c..bb7f973af 100644 --- a/src/opts.c +++ b/src/opts.c @@ -1,4 +1,4 @@ -// Copyright 2015-2024 The NATS Authors +// Copyright 2015-2025 The NATS Authors // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at @@ -989,6 +989,18 @@ natsOptions_SetMaxPendingBytes(natsOptions* opts, int64_t maxPending) return NATS_OK; } +natsStatus +natsOptions_SetProxyConnHandler(natsOptions* opts, natsProxyConnHandler proxyConnHandler) +{ + LOCK_AND_CHECK_OPTIONS(opts, 0) + + opts->proxyConnectCb = proxyConnHandler; + + UNLOCK_OPTS(opts); + + return NATS_OK; +} + natsStatus natsOptions_SetErrorHandler(natsOptions *opts, natsErrHandler errHandler, void *closure) diff --git a/test/list_test.txt b/test/list_test.txt index df12951c1..95518947c 100644 --- a/test/list_test.txt +++ b/test/list_test.txt @@ -220,6 +220,7 @@ _test(PingReconnect) _test(ProcessMsgArgs) _test(ProperFalloutAfterMaxAttempts) _test(ProperReconnectDelay) +_test(ProxyConnectCb) _test(PublishMsg) _test(PubSubWithReply) _test(QueueSubscriber) diff --git a/test/test.c b/test/test.c index cf86650d4..d8faadf1d 100644 --- a/test/test.c +++ b/test/test.c @@ -2705,6 +2705,12 @@ _dummyTokenHandler(void *closure) return "token"; } +static natsStatus +_dummyProxyConnHandler(char* host, int port, natsSock* fd) +{ + return NATS_SOCK_ERROR; +} + static void _dummyErrHandler(natsConnection *nc, natsSubscription *sub, natsStatus err, void *closure) @@ -2999,6 +3005,14 @@ void test_natsOptions(void) test("Set Max Pending Bytes : "); s = natsOptions_SetMaxPendingBytes(opts, 1000000); testCond((s == NATS_OK) && (opts->maxPendingBytes == 1000000)) + + test("Set Proxy Connection Handler: ") + s = natsOptions_SetProxyConnHandler(opts, _dummyProxyConnHandler); + testCond(s == NATS_OK && opts->proxyConnectCb == _dummyProxyConnHandler) + + test("Remove Proxy Connection Handler: ") + s = natsOptions_SetProxyConnHandler(opts, NULL); + testCond(s == NATS_OK && opts->proxyConnectCb == NULL) test("Set Error Handler: "); s = natsOptions_SetErrorHandler(opts, _dummyErrHandler, NULL); @@ -5952,6 +5966,80 @@ void test_ReconnectServerStats(void) _destroyDefaultThreadArgs(&args); } + +static natsStatus +_proxyConnectCb(char* host, int port, natsSock* pSock) +{ + natsStatus s = NATS_OK; + SOCKADDR_IN sockaddr_in; + + natsSock fd = socket(AF_INET, SOCK_STREAM, IPPROTO_TCP); + *pSock = fd; + if (fd == NATS_SOCK_INVALID) + return nats_setError(NATS_SYS_ERROR, "socket error: %d", NATS_SOCK_GET_ERROR); + + // In case of an existing proxy scenario we would use the proxy host and port as the socket address + // and send a connect request. If the connect request is successful (i.e. 200 Connection established), + // it is ready for use. + // For this test, we will just connect to the server directly. + + memset(&sockaddr_in, 0, sizeof(sockaddr_in)); + sockaddr_in.sin_family = AF_INET; + sockaddr_in.sin_port = htons((unsigned short)port); + inet_pton(AF_INET, host, &sockaddr_in.sin_addr); + + struct sockaddr* sockaddr = (struct sockaddr*)&sockaddr_in; + socklen_t socklen = (socklen_t)sizeof(sockaddr_in); + + s = natsSock_SetBlocking(fd, false); + if (s == NATS_OK) + { + int ret = connect(fd, sockaddr, socklen); + if (ret == NATS_SOCK_ERROR) + return nats_setDefaultError(NATS_NO_SERVER); + + s = natsSock_SetCommonTcpOptions(fd); + } + + return s; +} + +void test_ProxyConnectCb(void) +{ + natsStatus s; + natsConnection* nc = NULL; + natsOptions* opts = NULL; + natsPid serverPid = NATS_INVALID_PID; + struct threadArg args; + + test("ProxyConnectCb Fail: ") + + s = _createDefaultThreadArgsForCbTests(&args); + if (s != NATS_OK) + FAIL("Unable to setup test") + + serverPid = _startServer("nats://127.0.0.1:4222", NULL, true); + CHECK_SERVER_STARTED(serverPid) + + s = natsOptions_Create(&opts); + IFOK(s, natsOptions_SetURL(opts, NATS_DEFAULT_URL)) + IFOK(s, natsOptions_SetProxyConnHandler(opts, _dummyProxyConnHandler)) + + s = natsConnection_ConnectTo(&nc, NATS_DEFAULT_URL); + if (s == NATS_OK) + FAIL("Test not failed!") + + test("ProxyConnectCb OK: ") + + IFOK(s, natsOptions_SetProxyConnHandler(opts, _proxyConnectCb)) + s = natsConnection_ConnectTo(&nc, NATS_DEFAULT_URL); + + if (s != NATS_OK) + FAIL("Test failed!") + + _stopServer(serverPid); +} + static void _disconnectedCb(natsConnection *nc, void *closure) {