Skip to content

HTTP client: CONNECT tunnel support #1466

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
WaberZhuang wants to merge 1 commit into
base: main
Choose a base branch
from
Open

HTTP client: CONNECT tunnel support #1466

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ecosystem/test/test_oss_custom_headers.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -79,7 +79,7 @@ class OssCustomHeaderTest : public ::testing::Test {

ClientOptions make_opts() {
ClientOptions opts;
opts.endpoint = "oss-test.example.com";
opts.endpoint = "http://oss-test.example.com";
opts.bucket = "test-bucket";
opts.proxy = estring().appends("http://127.0.0.1:",
tcp_server->getsockname().port);
Expand Down
239 changes: 173 additions & 66 deletions net/http/client.cpp
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,6 +19,7 @@ limitations under the License.
#include <algorithm>
#include <random>
#include <photon/common/alog-stdstring.h>
#include <photon/common/estring.h>
#include <photon/common/iovector.h>
#include <photon/common/string_view.h>
#include <photon/net/socket.h>
Expand All @@ -37,9 +38,9 @@ class PooledDialer {
public:
net::TLSContext* tls_ctx = nullptr;
std::unique_ptr<ISocketClient> tcpsock;
std::unique_ptr<ISocketClient> tlssock;
std::unique_ptr<ISocketClient> udssock;
std::unique_ptr<Resolver> resolver;
std::unique_ptr<ISocketPool> tcp_pool;
photon::mutex init_mtx;
bool initialized = false;
bool tls_ctx_ownership = false;
Expand All @@ -61,73 +62,172 @@ class PooledDialer {
tls_ctx = new_tls_context(nullptr, nullptr, nullptr);
tls_ctx->set_verify_mode(VerifyMode::PEER); // act like curl
}
auto tcp_cli = new_tcp_socket_client(src_ips.data(), src_ips.size());
auto tls_cli = new_tls_client(tls_ctx, new_tcp_socket_client(src_ips.data(), src_ips.size()), true);
tcpsock.reset(new_tcp_socket_pool(tcp_cli, -1, true));
tlssock.reset(new_tcp_socket_pool(tls_cli, -1, true));
tcpsock.reset(new_tcp_socket_client(src_ips.data(), src_ips.size()));
udssock.reset(new_uds_client());
tcp_pool.reset(new_tcp_socket_pool(nullptr, -1ULL, false));
resolver.reset(new_default_resolver(kDNSCacheLife));
initialized = true;
return 0;
}

void at_photon_fini() {
tcp_pool.reset();
resolver.reset();
udssock.reset();
tlssock.reset();
tcpsock.reset();
if (tls_ctx_ownership)
delete tls_ctx;
initialized = false;
tls_ctx_ownership = false;
}

ISocketStream* dial(std::string_view host, uint16_t port, bool secure,
uint64_t timeout = -1ULL);
// Single entry point: choose dial method based on Operation proxy config.
ISocketStream* dial(Client::Operation* op, Timeout tmo);

template <typename T>
ISocketStream* dial(const T& x, uint64_t timeout = -1ULL) {
return dial(x.host_no_port(), x.port(), x.secure(), timeout);
ISocketStream* dial(std::string_view host, uint16_t port, bool secure, uint64_t timeout);

private:
// Key format for pool connection grouping:
// direct/proxy: <host>:<port>:<0|1> (0=TCP, 1=TLS)
// tunnel: <proxy_host>:<proxy_port>:<s|p>:<target_host>:<target_port>[:<auth>]
static std::string make_key(std::string_view host, uint16_t port, bool secure) {
return estring().appends(host, ":", port, ":", secure ? "1" : "0");
}

ISocketStream* dial(std::string_view uds_path, uint64_t timeout = -1ULL);
};
static std::string make_tunnel_key(const StoredURL& proxy, std::string_view target_host,
uint16_t target_port, const Headers* headers) {
auto auth = headers ? headers->get_value("Proxy-Authorization") : std::string_view{};

@lihuiba lihuiba Jul 19, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

auto auth = estring::make_conditional_cat_list(headers, ":", headers->get_value("Proxy-Authorization"));

auto phost = proxy.host_no_port();

@lihuiba lihuiba Jul 19, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

no need phost variable

return estring().appends(phost, ":", proxy.port(), ":", proxy.secure() ? "s" : "p",
":", target_host, ":", target_port,
estring::make_conditional_cat_list(!auth.empty(), ":", auth));
}

ISocketStream* PooledDialer::dial(std::string_view host, uint16_t port, bool secure, uint64_t timeout) {
LOG_DEBUG("Dialing to `:`", host, port);
auto ipaddr = resolver->resolve(host);
if (ipaddr.undefined()) {
LOG_ERROR_RETURN(ENOENT, nullptr, "DNS resolve failed, name = `", host)
// Factory: DNS resolve -> TCP connect -> optional TLS wrap with SNI.
// Discards the resolved IP from cache on connection failure.
ISocketStream* make_stream(std::string_view host, uint16_t port, bool secure, uint64_t timeout) {
auto ip = resolver->resolve(host);
if (ip.undefined())
LOG_ERROR_RETURN(ENOENT, nullptr, "DNS resolve failed, name=`", host);
EndPoint ep(ip, port);
tcpsock->timeout(timeout);
auto stream = tcpsock->connect(ep);
if (!stream) {
resolver->discard_cache(host, ip);
LOG_ERROR_RETURN(0, nullptr, "connection failed, ep=` host=` secure=`", ep, host, secure);
}
if (secure) {
stream = new_tls_stream(tls_ctx, stream, SecurityRole::Client, true);
if (!stream)
LOG_ERRNO_RETURN(0, nullptr, "TLS handshake failed, host=`", host);
tls_stream_set_hostname(stream, std::string(host).c_str());
}
LOG_DEBUG("Connected ` ssl:` host:`", ep, secure, host);
return stream;
}

EndPoint ep(ipaddr, port);
LOG_DEBUG("Connecting ` ssl: `", ep, secure);
ISocketStream *sock = nullptr;
if (secure) {
tlssock->timeout(timeout);
sock = tlssock->connect(ep);
tls_stream_set_hostname(sock, estring_view(host).extract_c_str());
} else {
// Factory: full CONNECT tunnel handshake sequence.
// DNS resolve proxy -> TCP -> [TLS to proxy] -> CONNECT handshake -> TLS to target.
ISocketStream* make_tunnel_stream(const StoredURL& proxy_url, std::string_view target_host,
uint16_t target_port, const Headers* headers, uint64_t timeout) {
auto proxy_ip = resolver->resolve(proxy_url.host_no_port());
if (proxy_ip.undefined())
LOG_ERROR_RETURN(ENOENT, nullptr, "CONNECT tunnel: DNS resolve failed for proxy `", proxy_url.host_no_port());
EndPoint proxy_ep(proxy_ip, proxy_url.port());
tcpsock->timeout(timeout);
sock = tcpsock->connect(ep);
auto stream = tcpsock->connect(proxy_ep);
if (!stream) {
resolver->discard_cache(proxy_url.host_no_port(), proxy_ip);
LOG_ERRNO_RETURN(0, nullptr, "CONNECT tunnel: failed to connect to proxy");
}
stream->timeout(timeout);
if (proxy_url.secure()) {
auto tls = new_tls_stream(tls_ctx, stream, SecurityRole::Client, true);
if (!tls) {
delete stream;

@lihuiba lihuiba Jul 19, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

make use of DEFER macro to delete it in case of error

LOG_ERRNO_RETURN(0, nullptr, "CONNECT tunnel: TLS to proxy failed");
}
stream = tls;
tls_stream_set_hostname(stream, std::string(proxy_url.host_no_port()).c_str());
}
if (do_connect_handshake(stream, target_host, target_port, headers, timeout) != 0) {
delete stream;
return nullptr;
}
auto tls = new_tls_stream(tls_ctx, stream, SecurityRole::Client, true);
if (!tls) {
delete stream;
LOG_ERRNO_RETURN(0, nullptr, "CONNECT tunnel: TLS to target failed");
}
stream = tls;
tls_stream_set_hostname(stream, std::string(target_host).c_str());
return stream;
}

// Send CONNECT request and verify 2xx response.
int do_connect_handshake(ISocketStream* stream, std::string_view target_host,
uint16_t target_port, const Headers* headers, uint64_t timeout) {
char buf[4096];
estring authority;
authority.appends(target_host, ":", target_port);
Request req(buf, (uint16_t)sizeof(buf), Verb::CONNECT, authority);
if (headers) {
for (auto it = headers->begin(); it != headers->end(); ++it)
req.headers.insert(it.first(), it.second());

@lihuiba lihuiba Jul 19, 2026

Copy link
Copy Markdown
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

meger_headers?

}
if (req.send_header(stream) < 0)
LOG_ERRNO_RETURN(0, -1, "CONNECT tunnel: failed to send CONNECT request");

// Response needs room for receive_bytes' MAX_TRANSFER_BYTES + RESERVED_INDEX_SIZE.
char rbuf[8192];
Response resp(rbuf, (uint16_t)sizeof(rbuf));
resp.reset(stream, false);
if (resp.receive_header(timeout) != 0)
LOG_ERRNO_RETURN(0, -1, "CONNECT tunnel: failed to read proxy response");
auto sc = resp.status_code();
if (sc < 200 || sc >= 300)
LOG_ERROR_RETURN(EINVAL, -1, "CONNECT tunnel: proxy returned status `", sc);
return 0;
}
};

ISocketStream* PooledDialer::dial(Client::Operation* op, Timeout tmo) {
auto* proxy = op->get_active_proxy();

// HTTPS target + proxy: CONNECT tunnel.
if (proxy && op->req.secure()) {
auto key = make_tunnel_key(*proxy, op->req.host_no_port(), op->req.port(), &op->proxy_header);
return tcp_pool->connect(key, [&]() -> ISocketStream* {
return make_tunnel_stream(*proxy, op->req.host_no_port(), op->req.port(), &op->proxy_header, tmo.timeout());
});
}
if (sock) {
LOG_DEBUG("Connected ` ", ep, VALUE(host), VALUE(secure));
return sock;

// HTTP proxy: connect to proxy endpoint.
if (proxy) {
auto key = make_key(proxy->host_no_port(), proxy->port(), proxy->secure());
return tcp_pool->connect(key, [&]() -> ISocketStream* {
return make_stream(proxy->host_no_port(), proxy->port(), proxy->secure(), tmo.timeout());
});
}
LOG_ERROR("connection failed, ssl : ` ep : ` host : `", secure, ep, host);
// When failed, remove resolved result from dns cache so that following retries can try
// different ips.
resolver->discard_cache(host, ipaddr);
return nullptr;

// Unix domain socket.
if (!op->uds_path.empty()) {
udssock->timeout(tmo.timeout());
return udssock->connect(op->uds_path.data());
}

// Direct TCP/TLS connection.
auto key = make_key(op->req.host_no_port(), op->req.port(), op->req.secure());
return tcp_pool->connect(key, [&]() -> ISocketStream* {
return make_stream(op->req.host_no_port(), op->req.port(), op->req.secure(), tmo.timeout());
});
}

ISocketStream* PooledDialer::dial(std::string_view uds_path, uint64_t timeout) {
udssock->timeout(timeout);
auto stream = udssock->connect(uds_path.data());
if (!stream)
LOG_ERRNO_RETURN(0, nullptr, "failed to dial to unix socket `", uds_path);
return stream;
ISocketStream* PooledDialer::dial(std::string_view host, uint16_t port, bool secure, uint64_t timeout) {
auto key = make_key(host, port, secure);
return tcp_pool->connect(key, [&]() -> ISocketStream* {
return make_stream(host, port, secure, timeout);
});
}

constexpr uint64_t code3xx() { return 0; }
Expand All @@ -141,19 +241,6 @@ constexpr static std::bitset<10>

static constexpr size_t kMinimalHeadersSize = 8 * 1024 - 1;

void Client::set_proxy(std::string_view proxy) {
m_proxy_url.from_string(proxy);
m_proxy = true;
auto ui = m_proxy_url.user_passwd();
if (!ui.empty()) {
std::string encoded;
Base64Encode(ui, encoded);
m_proxy_auth = "Basic " + encoded;
} else {
m_proxy_auth.clear();
}
}

enum RoundtripStatus {
ROUNDTRIP_SUCCESS,
ROUNDTRIP_FAILED,
Expand All @@ -166,6 +253,7 @@ enum RoundtripStatus {
class ClientImpl : public Client {
public:
CommonHeaders<> m_common_headers;
CommonHeaders<> m_proxy_headers;
TLSContext *m_tls_ctx;
ICookieJar *m_cookie_jar;
ClientImpl(ICookieJar *cookie_jar, TLSContext *tls_ctx) :
Expand Down Expand Up @@ -202,7 +290,7 @@ class ClientImpl : public Client {
"invalid 3xx status code: ", op->status_code);
}

if (op->req.redirect(v, location, op->enable_proxy) < 0) {
if (op->req.redirect(v, location, op->get_active_proxy() != nullptr) < 0) {
LOG_ERRNO_RETURN(0, ROUNDTRIP_FAILED, "redirect failed");
}
return ROUNDTRIP_REDIRECT;
Expand All @@ -213,15 +301,7 @@ class ClientImpl : public Client {
if (tmo.timeout() == 0)
LOG_ERROR_RETURN(ETIMEDOUT, ROUNDTRIP_FAILED, "connection timedout");
auto &req = op->req;
ISocketStream* s;
if (op->enable_proxy && !op->proxy_url.empty())
s = get_dialer().dial(op->proxy_url, tmo.timeout());
else if (op->enable_proxy && !m_proxy_url.empty())
s = get_dialer().dial(m_proxy_url, tmo.timeout());
else if (!op->uds_path.empty())
s = get_dialer().dial(op->uds_path, tmo.timeout());
else
s = get_dialer().dial(req, tmo.timeout());
ISocketStream* s = get_dialer().dial(op, tmo);
if (!s) {
if (errno == ECONNREFUSED || errno == ENOENT) {
LOG_ERROR_RETURN(0, ROUNDTRIP_FAST_RETRY, "connection refused")
Expand Down Expand Up @@ -304,10 +384,33 @@ class ClientImpl : public Client {
op->req.headers.insert("User-Agent", m_user_agent.empty() ? std::string_view(USERAGENT)
: std::string_view(m_user_agent));
op->req.headers.insert("Connection", "keep-alive");
if (op->enable_proxy && !m_proxy_auth.empty())
op->req.headers.insert("Proxy-Authorization", m_proxy_auth);
if (m_cookie_jar && m_cookie_jar->set_cookies_to_headers(&op->req) != 0)
LOG_ERROR_RETURN(0, -1, "set_cookies_to_headers failed");

// Proxy header assembly: op proxy_header <- client proxy_headers <- URL auth
auto proxy = op->get_active_proxy();
if (proxy) {
op->proxy_header.merge(m_proxy_headers);
if (!proxy->user_passwd().empty()) {
std::string encoded;
Base64Encode(proxy->user_passwd(), encoded);
op->proxy_header.insert("Proxy-Authorization", "Basic " + encoded);
}
}

// Rebuild request line if necessary (absolute URI for HTTP proxy)
auto need_abs = proxy && !op->req.secure();
auto is_abs = what_protocol(estring_view(op->req.target())) != 0;
if (need_abs != is_abs) {
op->req.redirect(op->req.verb(), op->req.target(), need_abs);
}

// For HTTP/1.1 proxy: merge proxy_header into req.headers (sent with the request).
// For HTTPS proxy: proxy_header stays separate (used by CONNECT handshake only).
if (proxy && !op->req.secure()) {
op->req.headers.merge(op->proxy_header);
}

Timeout tmo(std::min(op->timeout.timeout(), m_timeout));
int retry = 0, followed = 0, ret = 0;
uint64_t sleep_interval = 0;
Expand Down Expand Up @@ -346,6 +449,10 @@ class ClientImpl : public Client {
CommonHeaders<>* common_headers() override {
return &m_common_headers;
}

CommonHeaders<>* proxy_headers() override {
return &m_proxy_headers;
}
};

Client* new_http_client(ICookieJar *cookie_jar, TLSContext *tls_ctx) {
Expand Down
Loading
Loading