1- import type { Buffer } from 'node:buffer' ;
1+ import { Buffer } from 'node:buffer' ;
22import type dns from 'node:dns' ;
33import type { EventEmitter } from 'node:events' ;
4- import http from 'node:http' ;
5- import https from 'node:https' ;
4+ import type http from 'node:http' ;
5+ import type https from 'node:https' ;
6+ import net from 'node:net' ;
7+ import tls from 'node:tls' ;
68import type { URL } from 'node:url' ;
79
810import type { Socket } from './socket.js' ;
@@ -11,15 +13,6 @@ import type { SocketWithPreviousStats } from './utils/count_target_bytes.js';
1113import { countTargetBytes } from './utils/count_target_bytes.js' ;
1214import { getBasicAuthorizationHeader } from './utils/get_basic.js' ;
1315
14- interface Options {
15- method : string ;
16- headers : Record < string , string > ;
17- path ?: string ;
18- localAddress ?: string ;
19- family ?: number ;
20- lookup ?: typeof dns [ 'lookup' ] ;
21- }
22-
2316export interface HandlerOpts {
2417 upstreamProxyUrlParsed : URL ;
2518 ignoreUpstreamProxyCertificate : boolean ;
@@ -44,6 +37,13 @@ interface ChainOpts {
4437 * Passes the traffic to upstream HTTP proxy server.
4538 * Client -> Apify -> Upstream -> Web
4639 * Client <- Apify <- Upstream <- Web
40+ *
41+ * Uses raw TCP/TLS sockets to establish the upstream CONNECT tunnel rather
42+ * than `http.request().on('connect', ...)`. The latter is implemented on top
43+ * of `fetch()` in Bun 1.3 and (a) rejects non-URL CONNECT paths like `:443`
44+ * with "fetch() URL is invalid", (b) silently swallows the 407/590-class
45+ * upstream responses tests rely on. Speaking CONNECT directly over a socket
46+ * sidesteps both quirks without changing the behaviour on Node.
4747 */
4848export const chain = (
4949 {
@@ -67,151 +67,183 @@ export const chain = (
6767 }
6868
6969 const { proxyChainId } = sourceSocket ;
70-
7170 const { upstreamProxyUrlParsed : proxy , customTag } = handlerOpts ;
7271
73- const options : Options = {
74- method : 'CONNECT' ,
75- path : request . url ,
76- headers : {
77- host : request . url ! ,
78- } ,
72+ const isHttps = proxy . protocol === 'https:' ;
73+ const proxyHost = proxy . hostname ;
74+ const proxyPort = Number ( proxy . port ) || ( isHttps ? 443 : 80 ) ;
75+
76+ let connectRequest = `CONNECT ${ request . url } HTTP/1.1\r\nHost: ${ request . url } \r\n` ;
77+ if ( proxy . username || proxy . password ) {
78+ connectRequest += `Proxy-Authorization: ${ getBasicAuthorizationHeader ( proxy ) } \r\n` ;
79+ }
80+ connectRequest += '\r\n' ;
81+
82+ const socketOptions : net . TcpNetConnectOpts = {
83+ host : proxyHost ,
84+ port : proxyPort ,
7985 localAddress : handlerOpts . localAddress ,
80- family : handlerOpts . ipFamily ,
86+ family : handlerOpts . ipFamily as 4 | 6 | undefined ,
8187 lookup : handlerOpts . dnsLookup ,
8288 } ;
8389
84- if ( proxy . username || proxy . password ) {
85- options . headers [ 'proxy-authorization' ] = getBasicAuthorizationHeader ( proxy ) ;
86- }
90+ let targetSocket : net . Socket ;
8791
88- const client = proxy . protocol === 'https:'
89- ? https . request ( proxy . origin , {
90- ...options ,
91- rejectUnauthorized : ! handlerOpts . ignoreUpstreamProxyCertificate ,
92- agent : handlerOpts . httpsAgent ,
93- } )
94- : http . request ( proxy . origin , {
95- ...options ,
96- agent : handlerOpts . httpAgent ,
97- } ) ;
98-
99- client . once ( 'socket' , ( targetSocket : SocketWithPreviousStats ) => {
100- // Socket can be re-used by multiple requests.
101- // That's why we need to track the previous stats.
102- targetSocket . previousBytesRead = targetSocket . bytesRead ;
103- targetSocket . previousBytesWritten = targetSocket . bytesWritten ;
104- countTargetBytes ( sourceSocket , targetSocket ) ;
105- } ) ;
92+ const onPreConnectError = ( error : NodeJS . ErrnoException ) : void => {
93+ server . log ( proxyChainId , `Failed to connect to upstream proxy: ${ error . stack } ` ) ;
10694
107- client . on ( 'connect' , ( response , targetSocket , clientHead ) => {
108- if ( sourceSocket . readyState !== 'open' ) {
109- // Sanity check, should never reach.
110- targetSocket . destroy ( ) ;
111- return ;
95+ if ( sourceSocket . readyState === 'open' ) {
96+ if ( isPlain ) {
97+ sourceSocket . end ( ) ;
98+ } else {
99+ const statusCode = errorCodeToStatusCode [ error . code ! ] ?? badGatewayStatusCodes . GENERIC_ERROR ;
100+ const response = createCustomStatusHttpResponse ( statusCode , error . code ?? 'Upstream Closed Early' ) ;
101+ sourceSocket . end ( response ) ;
102+ }
112103 }
104+ } ;
113105
114- targetSocket . on ( 'error' , ( error ) => {
115- server . log ( proxyChainId , `Chain Destination Socket Error: ${ error . stack } ` ) ;
106+ const onProxyConnected = ( ) : void => {
107+ targetSocket . write ( connectRequest ) ;
116108
117- sourceSocket . destroy ( ) ;
118- } ) ;
109+ let responseBuffer = Buffer . alloc ( 0 ) ;
119110
120- sourceSocket . on ( 'error' , ( error ) => {
121- server . log ( proxyChainId , `Chain Source Socket Error: ${ error . stack } ` ) ;
111+ const onData = ( chunk : Buffer ) : void => {
112+ responseBuffer = Buffer . concat ( [ responseBuffer , chunk ] ) ;
122113
123- targetSocket . destroy ( ) ;
124- } ) ;
114+ const headerEnd = responseBuffer . indexOf ( '\r\n\r\n' ) ;
115+ if ( headerEnd === - 1 ) return ;
125116
126- if ( response . statusCode !== 200 ) {
127- server . log ( proxyChainId , `Failed to authenticate upstream proxy: ${ response . statusCode } ` ) ;
117+ targetSocket . removeListener ( 'data' , onData ) ;
128118
129- if ( isPlain ) {
130- sourceSocket . end ( ) ;
131- } else {
132- const { statusCode } = response ;
133- const status = statusCode === 401 || statusCode === 407
134- ? badGatewayStatusCodes . AUTH_FAILED
135- : badGatewayStatusCodes . NON_200 ;
119+ const headerStr = responseBuffer . subarray ( 0 , headerEnd ) . toString ( ) ;
120+ const remaining = responseBuffer . subarray ( headerEnd + 4 ) ;
121+
122+ const statusMatch = headerStr . match ( / ^ H T T P \/ \d + (?: \. \d + ) ? ( \d + ) (?: ( .* ) ) ? / ) ;
123+ const statusCode = statusMatch ? parseInt ( statusMatch [ 1 ] , 10 ) : 0 ;
124+ const statusMessage = statusMatch ? ( statusMatch [ 2 ] || '' ) : '' ;
125+
126+ const headers : Record < string , string > = { } ;
127+ const rawHeaders : string [ ] = [ ] ;
128+ for ( const line of headerStr . split ( '\r\n' ) . slice ( 1 ) ) {
129+ if ( ! line ) continue ;
130+ const colonIdx = line . indexOf ( ':' ) ;
131+ if ( colonIdx > 0 ) {
132+ const name = line . slice ( 0 , colonIdx ) . trim ( ) ;
133+ const value = line . slice ( colonIdx + 1 ) . trim ( ) ;
134+ headers [ name . toLowerCase ( ) ] = value ;
135+ rawHeaders . push ( name , value ) ;
136+ }
137+ }
138+
139+ const response = { statusCode, statusMessage, headers, rawHeaders } as unknown as http . IncomingMessage ;
140+
141+ if ( sourceSocket . readyState !== 'open' ) {
142+ targetSocket . destroy ( ) ;
143+ return ;
144+ }
145+
146+ targetSocket . removeListener ( 'error' , onPreConnectError ) ;
147+
148+ targetSocket . on ( 'error' , ( error ) => {
149+ server . log ( proxyChainId , `Chain Destination Socket Error: ${ error . stack } ` ) ;
150+ sourceSocket . destroy ( ) ;
151+ } ) ;
136152
137- sourceSocket . end ( createCustomStatusHttpResponse ( status , `UPSTREAM${ statusCode } ` ) ) ;
153+ sourceSocket . on ( 'error' , ( error ) => {
154+ server . log ( proxyChainId , `Chain Source Socket Error: ${ error . stack } ` ) ;
155+ targetSocket . destroy ( ) ;
156+ } ) ;
157+
158+ if ( statusCode !== 200 ) {
159+ server . log ( proxyChainId , `Failed to authenticate upstream proxy: ${ statusCode } ` ) ;
160+
161+ if ( isPlain ) {
162+ sourceSocket . end ( ) ;
163+ } else {
164+ const status = statusCode === 401 || statusCode === 407
165+ ? badGatewayStatusCodes . AUTH_FAILED
166+ : badGatewayStatusCodes . NON_200 ;
167+
168+ sourceSocket . end ( createCustomStatusHttpResponse ( status , `UPSTREAM${ statusCode } ` ) ) ;
169+ }
170+
171+ targetSocket . end ( ) ;
172+
173+ server . emit ( 'tunnelConnectFailed' , {
174+ proxyChainId,
175+ response,
176+ customTag,
177+ socket : targetSocket ,
178+ head : remaining ,
179+ } ) ;
180+
181+ return ;
138182 }
139183
140- targetSocket . end ( ) ;
184+ if ( remaining . length > 0 ) {
185+ // See comment above re: pre-response CONNECT payload
186+ targetSocket . unshift ( remaining ) ;
187+ }
141188
142- server . emit ( 'tunnelConnectFailed ' , {
189+ server . emit ( 'tunnelConnectResponded ' , {
143190 proxyChainId,
144191 response,
145192 customTag,
146193 socket : targetSocket ,
147- head : clientHead ,
194+ head : remaining ,
148195 } ) ;
149196
150- return ;
151- }
152-
153- if ( clientHead . length > 0 ) {
154- // See comment above
155- targetSocket . unshift ( clientHead ) ;
156- }
197+ sourceSocket . write ( isPlain ? '' : `HTTP/1.1 200 Connection Established\r\n\r\n` ) ;
157198
158- server . emit ( 'tunnelConnectResponded' , {
159- proxyChainId,
160- response,
161- customTag,
162- socket : targetSocket ,
163- head : clientHead ,
164- } ) ;
199+ sourceSocket . pipe ( targetSocket ) ;
200+ targetSocket . pipe ( sourceSocket ) ;
165201
166- sourceSocket . write ( isPlain ? '' : `HTTP/1.1 200 Connection Established\r\n\r\n` ) ;
202+ // Once target socket closes forcibly, the source socket gets paused.
203+ // We need to enable flowing, otherwise the socket would remain open indefinitely.
204+ // Nothing would consume the data, we just want to close the socket.
205+ targetSocket . on ( 'close' , ( ) => {
206+ sourceSocket . resume ( ) ;
167207
168- sourceSocket . pipe ( targetSocket ) ;
169- targetSocket . pipe ( sourceSocket ) ;
208+ if ( sourceSocket . writable ) {
209+ sourceSocket . end ( ) ;
210+ }
211+ } ) ;
170212
171- // Once target socket closes forcibly, the source socket gets paused.
172- // We need to enable flowing, otherwise the socket would remain open indefinitely.
173- // Nothing would consume the data, we just want to close the socket.
174- targetSocket . on ( 'close' , ( ) => {
175- sourceSocket . resume ( ) ;
213+ // Same here.
214+ sourceSocket . on ( 'close' , ( ) => {
215+ targetSocket . resume ( ) ;
176216
177- if ( sourceSocket . writable ) {
178- sourceSocket . end ( ) ;
179- }
180- } ) ;
217+ if ( targetSocket . writable ) {
218+ targetSocket . end ( ) ;
219+ }
220+ } ) ;
221+ } ;
181222
182- // Same here.
183- sourceSocket . on ( 'close' , ( ) => {
184- targetSocket . resume ( ) ;
223+ targetSocket . on ( 'data' , onData ) ;
224+ } ;
185225
186- if ( targetSocket . writable ) {
187- targetSocket . end ( ) ;
188- }
189- } ) ;
190- } ) ;
226+ if ( isHttps ) {
227+ targetSocket = tls . connect ( {
228+ ...socketOptions ,
229+ rejectUnauthorized : ! handlerOpts . ignoreUpstreamProxyCertificate ,
230+ } , onProxyConnected ) ;
231+ } else {
232+ targetSocket = net . createConnection ( socketOptions , onProxyConnected ) ;
233+ }
191234
192- client . on ( 'error' , ( error : NodeJS . ErrnoException ) => {
193- server . log ( proxyChainId , `Failed to connect to upstream proxy: ${ error . stack } ` ) ;
235+ ( targetSocket as SocketWithPreviousStats ) . previousBytesRead = 0 ;
236+ ( targetSocket as SocketWithPreviousStats ) . previousBytesWritten = 0 ;
237+ countTargetBytes ( sourceSocket , targetSocket ) ;
194238
195- // The end socket may get connected after the client to proxy one gets disconnected.
196- if ( sourceSocket . readyState === 'open' ) {
197- if ( isPlain ) {
198- sourceSocket . end ( ) ;
199- } else {
200- const statusCode = errorCodeToStatusCode [ error . code ! ] ?? badGatewayStatusCodes . GENERIC_ERROR ;
201- const response = createCustomStatusHttpResponse ( statusCode , error . code ?? 'Upstream Closed Early' ) ;
202- sourceSocket . end ( response ) ;
203- }
204- }
205- } ) ;
239+ targetSocket . on ( 'error' , onPreConnectError ) ;
206240
207241 sourceSocket . on ( 'error' , ( ) => {
208- client . destroy ( ) ;
242+ targetSocket . destroy ( ) ;
209243 } ) ;
210244
211245 // In case the client ends the socket too early
212246 sourceSocket . on ( 'close' , ( ) => {
213- client . destroy ( ) ;
247+ targetSocket . destroy ( ) ;
214248 } ) ;
215-
216- client . end ( ) ;
217249} ;
0 commit comments