Skip to content

Commit 474c2ba

Browse files
WaberZhuangWaberZhuang
committed
HTTP client: CONNECT tunnel support
Signed-off-by: zhuangbowei.zbw <zhuangbowei.zbw@alibaba-inc.com>
1 parent 8fd3e56 commit 474c2ba

5 files changed

Lines changed: 937 additions & 172 deletions

File tree

‎ecosystem/test/test_oss_custom_headers.cpp‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -79,7 +79,7 @@ class OssCustomHeaderTest : public ::testing::Test {
7979

8080
ClientOptions make_opts() {
8181
ClientOptions opts;
82-
opts.endpoint = "oss-test.example.com";
82+
opts.endpoint = "http://oss-test.example.com";
8383
opts.bucket = "test-bucket";
8484
opts.proxy = estring().appends("http://127.0.0.1:",
8585
tcp_server->getsockname().port);

‎net/http/client.cpp‎

Lines changed: 197 additions & 66 deletions
Original file line numberDiff line numberDiff line change
@@ -19,6 +19,7 @@ limitations under the License.
1919
#include <algorithm>
2020
#include <random>
2121
#include <photon/common/alog-stdstring.h>
22+
#include <photon/common/estring.h>
2223
#include <photon/common/iovector.h>
2324
#include <photon/common/string_view.h>
2425
#include <photon/net/socket.h>
@@ -37,9 +38,9 @@ class PooledDialer {
3738
public:
3839
net::TLSContext* tls_ctx = nullptr;
3940
std::unique_ptr<ISocketClient> tcpsock;
40-
std::unique_ptr<ISocketClient> tlssock;
4141
std::unique_ptr<ISocketClient> udssock;
4242
std::unique_ptr<Resolver> resolver;
43+
std::unique_ptr<ISocketPool> tcp_pool;
4344
photon::mutex init_mtx;
4445
bool initialized = false;
4546
bool tls_ctx_ownership = false;
@@ -61,73 +62,196 @@ class PooledDialer {
6162
tls_ctx = new_tls_context(nullptr, nullptr, nullptr);
6263
tls_ctx->set_verify_mode(VerifyMode::PEER); // act like curl
6364
}
64-
auto tcp_cli = new_tcp_socket_client(src_ips.data(), src_ips.size());
65-
auto tls_cli = new_tls_client(tls_ctx, new_tcp_socket_client(src_ips.data(), src_ips.size()), true);
66-
tcpsock.reset(new_tcp_socket_pool(tcp_cli, -1, true));
67-
tlssock.reset(new_tcp_socket_pool(tls_cli, -1, true));
65+
tcpsock.reset(new_tcp_socket_client(src_ips.data(), src_ips.size()));
6866
udssock.reset(new_uds_client());
67+
tcp_pool.reset(new_tcp_socket_pool(nullptr, -1ULL, false));
6968
resolver.reset(new_default_resolver(kDNSCacheLife));
7069
initialized = true;
7170
return 0;
7271
}
7372

7473
void at_photon_fini() {
74+
tcp_pool.reset();
7575
resolver.reset();
7676
udssock.reset();
77-
tlssock.reset();
7877
tcpsock.reset();
7978
if (tls_ctx_ownership)
8079
delete tls_ctx;
8180
initialized = false;
8281
tls_ctx_ownership = false;
8382
}
8483

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

88-
template <typename T>
89-
ISocketStream* dial(const T& x, uint64_t timeout = -1ULL) {
90-
return dial(x.host_no_port(), x.port(), x.secure(), timeout);
87+
ISocketStream* dial(std::string_view host, uint16_t port, bool secure, uint64_t timeout);
88+
89+
private:
90+
// Key format for pool connection grouping:
91+
// direct/proxy: <host>:<port>:<0|1> (0=TCP, 1=TLS)
92+
// tunnel: <proxy_host>:<proxy_port>:<s|p>:<target_host>:<target_port>[:<auth>]
93+
static std::string make_key(std::string_view host, uint16_t port, bool secure) {
94+
return estring().appends(host, ":", port, ":", secure ? '1' : '0');
95+
}
96+
97+
static std::string make_tunnel_key(const StoredURL& proxy, std::string_view target_host,
98+
uint16_t target_port, const Headers* headers) {
99+
auto auth = headers ? headers->get_value("Proxy-Authorization") : std::string_view{};
100+
auto phost = proxy.host_no_port();
101+
estring key;
102+
key.appends(phost, ":", proxy.port(), ":", proxy.secure() ? 's' : 'p',
103+
":", target_host, ":", target_port);
104+
if (!auth.empty()) {
105+
key += ':';
106+
key.append(auth.data(), auth.size());
107+
}
108+
return std::string(std::move(key));
109+
}
110+
111+
// Factory: DNS resolve -> TCP connect -> optional TLS wrap with SNI.
112+
// Discards the resolved IP from cache on connection failure.
113+
ISocketStream* make_stream(std::string_view host, uint16_t port, bool secure, uint64_t timeout) {
114+
auto ip = resolver->resolve(host);
115+
if (ip.undefined())
116+
LOG_ERROR_RETURN(ENOENT, nullptr, "DNS resolve failed, name=`", host);
117+
EndPoint ep(ip, port);
118+
tcpsock->timeout(timeout);
119+
auto stream = tcpsock->connect(ep);
120+
if (!stream) {
121+
resolver->discard_cache(host, ip);
122+
LOG_ERROR_RETURN(0, nullptr, "connection failed, ep=` host=` secure=`", ep, host, secure);
123+
}
124+
if (secure) {
125+
stream = new_tls_stream(tls_ctx, stream, SecurityRole::Client, true);
126+
if (!stream)
127+
LOG_ERRNO_RETURN(0, nullptr, "TLS handshake failed, host=`", host);
128+
tls_stream_set_hostname(stream, std::string(host).c_str());
129+
}
130+
LOG_DEBUG("Connected ` ssl:` host:`", ep, secure, host);
131+
return stream;
132+
}
133+
134+
// Factory: full CONNECT tunnel handshake sequence.
135+
// DNS resolve proxy -> TCP -> [TLS to proxy] -> CONNECT handshake -> TLS to target.
136+
ISocketStream* make_tunnel_stream(const StoredURL& proxy_url, std::string_view target_host,
137+
uint16_t target_port, const Headers* headers) {
138+
auto proxy_ip = resolver->resolve(proxy_url.host_no_port());
139+
if (proxy_ip.undefined())
140+
LOG_ERROR_RETURN(ENOENT, nullptr, "CONNECT tunnel: DNS resolve failed for proxy `", proxy_url.host_no_port());
141+
EndPoint proxy_ep(proxy_ip, proxy_url.port());
142+
auto stream = tcpsock->connect(proxy_ep);
143+
if (!stream) {
144+
resolver->discard_cache(proxy_url.host_no_port(), proxy_ip);
145+
LOG_ERRNO_RETURN(0, nullptr, "CONNECT tunnel: failed to connect to proxy");
146+
}
147+
if (proxy_url.secure()) {
148+
stream = new_tls_stream(tls_ctx, stream, SecurityRole::Client, true);
149+
if (!stream)
150+
LOG_ERRNO_RETURN(0, nullptr, "CONNECT tunnel: TLS to proxy failed");
151+
tls_stream_set_hostname(stream, std::string(proxy_url.host_no_port()).c_str());
152+
}
153+
if (do_connect_handshake(stream, target_host, target_port, headers) != 0) {
154+
delete stream;
155+
return nullptr;
156+
}
157+
stream = new_tls_stream(tls_ctx, stream, SecurityRole::Client, true);
158+
if (!stream)
159+
LOG_ERRNO_RETURN(0, nullptr, "CONNECT tunnel: TLS to target failed");
160+
tls_stream_set_hostname(stream, std::string(target_host).c_str());
161+
return stream;
91162
}
92163

93-
ISocketStream* dial(std::string_view uds_path, uint64_t timeout = -1ULL);
164+
// Send CONNECT request and verify 2xx response.
165+
int do_connect_handshake(ISocketStream* stream, std::string_view target_host,
166+
uint16_t target_port, const Headers* headers) {
167+
char buf[4096];
168+
// CONNECT target is always host:port per RFC 7231 S4.3.6.
169+
int n = snprintf(buf, sizeof(buf), "CONNECT %.*s:%u HTTP/1.1\r\n",
170+
(int)target_host.size(), target_host.data(), target_port);
171+
if (n < 0 || n >= (int)sizeof(buf))
172+
LOG_ERROR_RETURN(ENOMEM, -1, "CONNECT request too long");
173+
n += snprintf(buf + n, sizeof(buf) - n, "Host: %.*s:%u\r\n",
174+
(int)target_host.size(), target_host.data(), target_port);
175+
if (n >= (int)sizeof(buf))
176+
LOG_ERROR_RETURN(ENOMEM, -1, "CONNECT request too long");
177+
if (headers) {
178+
for (auto it = headers->begin(); it != headers->end(); ++it) {
179+
auto k = it.first();
180+
auto v = it.second();
181+
if ((size_t)(n + k.size() + v.size() + 4) >= sizeof(buf))
182+
LOG_ERROR_RETURN(ENOMEM, -1, "CONNECT request too long");
183+
memcpy(buf + n, k.data(), k.size()); n += k.size();
184+
buf[n++] = ':'; buf[n++] = ' ';
185+
memcpy(buf + n, v.data(), v.size()); n += v.size();
186+
buf[n++] = '\r'; buf[n++] = '\n';
187+
}
188+
}
189+
if (n + 2 >= (int)sizeof(buf))
190+
LOG_ERROR_RETURN(ENOMEM, -1, "CONNECT request too long");
191+
buf[n++] = '\r'; buf[n++] = '\n'; buf[n] = '\0';
192+
if (stream->write(buf, n) != n)
193+
LOG_ERRNO_RETURN(0, -1, "CONNECT tunnel: failed to send CONNECT request");
194+
195+
size_t total = 0;
196+
while (total < sizeof(buf) - 1) {
197+
auto rc = stream->recv(buf + total, sizeof(buf) - 1 - total);
198+
if (rc <= 0)
199+
LOG_ERRNO_RETURN(0, -1, "CONNECT tunnel: failed to read proxy response");
200+
total += rc;
201+
buf[total] = '\0';
202+
if (strstr(buf, "\r\n\r\n"))
203+
break;
204+
}
205+
auto space = strchr(buf, ' ');
206+
if (!space)
207+
LOG_ERROR_RETURN(EINVAL, -1, "CONNECT tunnel: invalid proxy response");
208+
int status_code = atoi(space + 1);
209+
if (status_code < 200 || status_code >= 300) {
210+
auto end = strstr(buf, "\r\n\r\n");
211+
if (end) *end = '\0';
212+
LOG_ERROR_RETURN(EINVAL, -1, "CONNECT tunnel: proxy returned status `, response: `", status_code, buf);
213+
}
214+
return 0;
215+
}
94216
};
95217

96-
ISocketStream* PooledDialer::dial(std::string_view host, uint16_t port, bool secure, uint64_t timeout) {
97-
LOG_DEBUG("Dialing to `:`", host, port);
98-
auto ipaddr = resolver->resolve(host);
99-
if (ipaddr.undefined()) {
100-
LOG_ERROR_RETURN(ENOENT, nullptr, "DNS resolve failed, name = `", host)
218+
ISocketStream* PooledDialer::dial(Client::Operation* op, Timeout tmo) {
219+
auto* proxy = op->get_active_proxy();
220+
221+
// HTTPS target + proxy: CONNECT tunnel.
222+
if (proxy && op->req.secure()) {
223+
auto key = make_tunnel_key(*proxy, op->req.host_no_port(), op->req.port(), &op->proxy_header);
224+
return tcp_pool->connect(key, [&]() -> ISocketStream* {
225+
return make_tunnel_stream(*proxy, op->req.host_no_port(), op->req.port(), &op->proxy_header);
226+
});
101227
}
102228

103-
EndPoint ep(ipaddr, port);
104-
LOG_DEBUG("Connecting ` ssl: `", ep, secure);
105-
ISocketStream *sock = nullptr;
106-
if (secure) {
107-
tlssock->timeout(timeout);
108-
sock = tlssock->connect(ep);
109-
tls_stream_set_hostname(sock, estring_view(host).extract_c_str());
110-
} else {
111-
tcpsock->timeout(timeout);
112-
sock = tcpsock->connect(ep);
229+
// HTTP proxy: connect to proxy endpoint.
230+
if (proxy) {
231+
auto key = make_key(proxy->host_no_port(), proxy->port(), proxy->secure());
232+
return tcp_pool->connect(key, [&]() -> ISocketStream* {
233+
return make_stream(proxy->host_no_port(), proxy->port(), proxy->secure(), tmo.timeout());
234+
});
113235
}
114-
if (sock) {
115-
LOG_DEBUG("Connected ` ", ep, VALUE(host), VALUE(secure));
116-
return sock;
236+
237+
// Unix domain socket.
238+
if (!op->uds_path.empty()) {
239+
udssock->timeout(tmo.timeout());
240+
return udssock->connect(op->uds_path.data());
117241
}
118-
LOG_ERROR("connection failed, ssl : ` ep : ` host : `", secure, ep, host);
119-
// When failed, remove resolved result from dns cache so that following retries can try
120-
// different ips.
121-
resolver->discard_cache(host, ipaddr);
122-
return nullptr;
242+
243+
// Direct TCP/TLS connection.
244+
auto key = make_key(op->req.host_no_port(), op->req.port(), op->req.secure());
245+
return tcp_pool->connect(key, [&]() -> ISocketStream* {
246+
return make_stream(op->req.host_no_port(), op->req.port(), op->req.secure(), tmo.timeout());
247+
});
123248
}
124249

125-
ISocketStream* PooledDialer::dial(std::string_view uds_path, uint64_t timeout) {
126-
udssock->timeout(timeout);
127-
auto stream = udssock->connect(uds_path.data());
128-
if (!stream)
129-
LOG_ERRNO_RETURN(0, nullptr, "failed to dial to unix socket `", uds_path);
130-
return stream;
250+
ISocketStream* PooledDialer::dial(std::string_view host, uint16_t port, bool secure, uint64_t timeout) {
251+
auto key = make_key(host, port, secure);
252+
return tcp_pool->connect(key, [&]() -> ISocketStream* {
253+
return make_stream(host, port, secure, timeout);
254+
});
131255
}
132256

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

142266
static constexpr size_t kMinimalHeadersSize = 8 * 1024 - 1;
143267

144-
void Client::set_proxy(std::string_view proxy) {
145-
m_proxy_url.from_string(proxy);
146-
m_proxy = true;
147-
auto ui = m_proxy_url.user_passwd();
148-
if (!ui.empty()) {
149-
std::string encoded;
150-
Base64Encode(ui, encoded);
151-
m_proxy_auth = "Basic " + encoded;
152-
} else {
153-
m_proxy_auth.clear();
154-
}
155-
}
156-
157268
enum RoundtripStatus {
158269
ROUNDTRIP_SUCCESS,
159270
ROUNDTRIP_FAILED,
@@ -166,6 +277,7 @@ enum RoundtripStatus {
166277
class ClientImpl : public Client {
167278
public:
168279
CommonHeaders<> m_common_headers;
280+
CommonHeaders<> m_proxy_headers;
169281
TLSContext *m_tls_ctx;
170282
ICookieJar *m_cookie_jar;
171283
ClientImpl(ICookieJar *cookie_jar, TLSContext *tls_ctx) :
@@ -202,7 +314,7 @@ class ClientImpl : public Client {
202314
"invalid 3xx status code: ", op->status_code);
203315
}
204316

205-
if (op->req.redirect(v, location, op->enable_proxy) < 0) {
317+
if (op->req.redirect(v, location, op->get_active_proxy() != nullptr) < 0) {
206318
LOG_ERRNO_RETURN(0, ROUNDTRIP_FAILED, "redirect failed");
207319
}
208320
return ROUNDTRIP_REDIRECT;
@@ -213,15 +325,7 @@ class ClientImpl : public Client {
213325
if (tmo.timeout() == 0)
214326
LOG_ERROR_RETURN(ETIMEDOUT, ROUNDTRIP_FAILED, "connection timedout");
215327
auto &req = op->req;
216-
ISocketStream* s;
217-
if (op->enable_proxy && !op->proxy_url.empty())
218-
s = get_dialer().dial(op->proxy_url, tmo.timeout());
219-
else if (op->enable_proxy && !m_proxy_url.empty())
220-
s = get_dialer().dial(m_proxy_url, tmo.timeout());
221-
else if (!op->uds_path.empty())
222-
s = get_dialer().dial(op->uds_path, tmo.timeout());
223-
else
224-
s = get_dialer().dial(req, tmo.timeout());
328+
ISocketStream* s = get_dialer().dial(op, tmo);
225329
if (!s) {
226330
if (errno == ECONNREFUSED || errno == ENOENT) {
227331
LOG_ERROR_RETURN(0, ROUNDTRIP_FAST_RETRY, "connection refused")
@@ -304,10 +408,33 @@ class ClientImpl : public Client {
304408
op->req.headers.insert("User-Agent", m_user_agent.empty() ? std::string_view(USERAGENT)
305409
: std::string_view(m_user_agent));
306410
op->req.headers.insert("Connection", "keep-alive");
307-
if (op->enable_proxy && !m_proxy_auth.empty())
308-
op->req.headers.insert("Proxy-Authorization", m_proxy_auth);
309411
if (m_cookie_jar && m_cookie_jar->set_cookies_to_headers(&op->req) != 0)
310412
LOG_ERROR_RETURN(0, -1, "set_cookies_to_headers failed");
413+
414+
// Proxy header assembly: op proxy_header <- client proxy_headers <- URL auth
415+
auto proxy = op->get_active_proxy();
416+
if (proxy) {
417+
op->proxy_header.merge(m_proxy_headers);
418+
if (!proxy->user_passwd().empty()) {
419+
std::string encoded;
420+
Base64Encode(proxy->user_passwd(), encoded);
421+
op->proxy_header.insert("Proxy-Authorization", "Basic " + encoded);
422+
}
423+
}
424+
425+
// Rebuild request line if necessary (absolute URI for HTTP proxy)
426+
auto need_abs = proxy && !op->req.secure();
427+
auto is_abs = what_protocol(estring_view(op->req.target())) != 0;
428+
if (need_abs != is_abs) {
429+
op->req.redirect(op->req.verb(), op->req.target(), need_abs);
430+
}
431+
432+
// For HTTP/1.1 proxy: merge proxy_header into req.headers (sent with the request).
433+
// For HTTPS proxy: proxy_header stays separate (used by CONNECT handshake only).
434+
if (proxy && !op->req.secure()) {
435+
op->req.headers.merge(op->proxy_header);
436+
}
437+
311438
Timeout tmo(std::min(op->timeout.timeout(), m_timeout));
312439
int retry = 0, followed = 0, ret = 0;
313440
uint64_t sleep_interval = 0;
@@ -346,6 +473,10 @@ class ClientImpl : public Client {
346473
CommonHeaders<>* common_headers() override {
347474
return &m_common_headers;
348475
}
476+
477+
CommonHeaders<>* proxy_headers() override {
478+
return &m_proxy_headers;
479+
}
349480
};
350481

351482
Client* new_http_client(ICookieJar *cookie_jar, TLSContext *tls_ctx) {

0 commit comments

Comments
 (0)