-
Notifications
You must be signed in to change notification settings - Fork 145
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
94640e5
commit a586b3e
Showing
8 changed files
with
290 additions
and
27 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,128 @@ | ||
| import http from 'http'; | ||
| import net from 'net'; | ||
| import { Buffer } from 'buffer'; | ||
| import { URL } from 'url'; | ||
| import { EventEmitter } from 'events'; | ||
| import { SocksClient, SocksClientError, type SocksProxy } from 'socks'; | ||
| import { countTargetBytes } from './utils/count_target_bytes'; | ||
| import { Socket } from './socket'; | ||
| import { createCustomStatusHttpResponse, socksErrorMessageToStatusCode } from './statuses'; | ||
|
|
||
| export interface HandlerOpts { | ||
| upstreamProxyUrlParsed: URL; | ||
| customTag?: unknown; | ||
| } | ||
|
|
||
| interface ChainSocksOpts { | ||
| request: http.IncomingMessage, | ||
| sourceSocket: Socket; | ||
| head: Buffer; | ||
| server: EventEmitter & { log: (connectionId: unknown, str: string) => void }; | ||
| handlerOpts: HandlerOpts; | ||
| } | ||
|
|
||
| const socksProtocolToVersionNumber = (protocol: string): 4 | 5 => { | ||
| switch (protocol) { | ||
| case 'socks4:': | ||
| case 'socks4a:': | ||
| return 4; | ||
| default: | ||
| return 5; | ||
| } | ||
| }; | ||
|
|
||
| /** | ||
| * Client -> Apify (CONNECT) -> Upstream (SOCKS) -> Web | ||
| * Client <- Apify (CONNECT) <- Upstream (SOCKS) <- Web | ||
| */ | ||
| export const chainSocks = async ({ | ||
| request, | ||
| sourceSocket, | ||
| head, | ||
| server, | ||
| handlerOpts, | ||
| }: ChainSocksOpts): Promise<void> => { | ||
| const { proxyChainId } = sourceSocket; | ||
|
|
||
| const { hostname, port, username, password } = handlerOpts.upstreamProxyUrlParsed; | ||
|
|
||
| const proxy: SocksProxy = { | ||
| host: hostname, | ||
| port: Number(port), | ||
| type: socksProtocolToVersionNumber(handlerOpts.upstreamProxyUrlParsed.protocol), | ||
| userId: username, | ||
| password, | ||
| }; | ||
|
|
||
| if (head && head.length > 0) { | ||
| // HTTP/1.1 has no defined semantics when sending payload along with CONNECT and servers can reject the request. | ||
| // HTTP/2 only says that subsequent DATA frames must be transferred after HEADERS has been sent. | ||
| // HTTP/3 says that all DATA frames should be transferred (implies pre-HEADERS data). | ||
| // | ||
| // Let's go with the HTTP/3 behavior. | ||
| // There are also clients that send payload along with CONNECT to save milliseconds apparently. | ||
| // Beware of upstream proxy servers that send out valid CONNECT responses with diagnostic data such as IPs! | ||
| sourceSocket.unshift(head); | ||
| } | ||
|
|
||
| const url = new URL(`connect://${request.url}`); | ||
| const destination = { | ||
| port: Number(url.port), | ||
| host: url.hostname, | ||
| }; | ||
|
|
||
| let targetSocket: net.Socket; | ||
|
|
||
| try { | ||
| const client = await SocksClient.createConnection({ | ||
| proxy, | ||
| command: 'connect', | ||
| destination, | ||
| }); | ||
| targetSocket = client.socket; | ||
|
|
||
| sourceSocket.write(`HTTP/1.1 200 Connection Established\r\n\r\n`); | ||
| } catch (error) { | ||
| const socksError = error as SocksClientError; | ||
| server.log(proxyChainId, `Failed to connect to upstream SOCKS proxy ${socksError.stack}`); | ||
| sourceSocket.end(createCustomStatusHttpResponse(socksErrorMessageToStatusCode(socksError.message), socksError.message)); | ||
| return; | ||
| } | ||
|
|
||
| countTargetBytes(sourceSocket, targetSocket); | ||
|
|
||
| sourceSocket.pipe(targetSocket); | ||
| targetSocket.pipe(sourceSocket); | ||
|
|
||
| // Once target socket closes forcibly, the source socket gets paused. | ||
| // We need to enable flowing, otherwise the socket would remain open indefinitely. | ||
| // Nothing would consume the data, we just want to close the socket. | ||
| targetSocket.on('close', () => { | ||
| sourceSocket.resume(); | ||
|
|
||
| if (sourceSocket.writable) { | ||
| sourceSocket.end(); | ||
| } | ||
| }); | ||
|
|
||
| // Same here. | ||
| sourceSocket.on('close', () => { | ||
| targetSocket.resume(); | ||
|
|
||
| if (targetSocket.writable) { | ||
| targetSocket.end(); | ||
| } | ||
| }); | ||
|
|
||
| targetSocket.on('error', (error) => { | ||
| server.log(proxyChainId, `Chain SOCKS Destination Socket Error: ${error.stack}`); | ||
|
|
||
| sourceSocket.destroy(); | ||
| }); | ||
|
|
||
| sourceSocket.on('error', (error) => { | ||
| server.log(proxyChainId, `Chain SOCKS Source Socket Error: ${error.stack}`); | ||
|
|
||
| targetSocket.destroy(); | ||
| }); | ||
| }; |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,102 @@ | ||
| import http from 'http'; | ||
| import stream from 'stream'; | ||
| import util from 'util'; | ||
| import { URL } from 'url'; | ||
| import { SocksProxyAgent } from 'socks-proxy-agent'; | ||
| import { validHeadersOnly } from './utils/valid_headers_only'; | ||
| import { countTargetBytes } from './utils/count_target_bytes'; | ||
| import { badGatewayStatusCodes, errorCodeToStatusCode } from './statuses'; | ||
|
|
||
| const pipeline = util.promisify(stream.pipeline); | ||
|
|
||
| interface Options { | ||
| method: string; | ||
| headers: string[]; | ||
| insecureHTTPParser: boolean; | ||
| path?: string; | ||
| localAddress?: string; | ||
| agent: http.Agent; | ||
| } | ||
|
|
||
| export interface HandlerOpts { | ||
| upstreamProxyUrlParsed: URL; | ||
| localAddress?: string; | ||
| } | ||
|
|
||
| /** | ||
| * ``` | ||
| * Client -> Apify (HTTP) -> Upstream (SOCKS) -> Web | ||
| * Client <- Apify (HTTP) <- Upstream (SOCKS) <- Web | ||
| * ``` | ||
| */ | ||
| export const forwardSocks = async ( | ||
| request: http.IncomingMessage, | ||
| response: http.ServerResponse, | ||
| handlerOpts: HandlerOpts, | ||
| // eslint-disable-next-line no-async-promise-executor | ||
| ): Promise<void> => new Promise(async (resolve, reject) => { | ||
| const agent = new SocksProxyAgent(handlerOpts.upstreamProxyUrlParsed); | ||
|
|
||
| const options: Options = { | ||
| method: request.method!, | ||
| headers: validHeadersOnly(request.rawHeaders), | ||
| insecureHTTPParser: true, | ||
| localAddress: handlerOpts.localAddress, | ||
| agent, | ||
| }; | ||
|
|
||
| // Only handling "http" here - since everything else is handeled by tunnelSocks. | ||
| // We have to force cast `options` because @types/node doesn't support an array. | ||
| const client = http.request(request.url!, options as unknown as http.ClientRequestArgs, async (clientResponse) => { | ||
| try { | ||
| // This is necessary to prevent Node.js throwing an error | ||
| let statusCode = clientResponse.statusCode!; | ||
| if (statusCode < 100 || statusCode > 999) { | ||
| statusCode = badGatewayStatusCodes.STATUS_CODE_OUT_OF_RANGE; | ||
| } | ||
|
|
||
| // 407 is handled separately | ||
| if (clientResponse.statusCode === 407) { | ||
| reject(new Error('407 Proxy Authentication Required')); | ||
| return; | ||
| } | ||
|
|
||
| response.writeHead( | ||
| statusCode, | ||
| clientResponse.statusMessage, | ||
| validHeadersOnly(clientResponse.rawHeaders), | ||
| ); | ||
|
|
||
| // `pipeline` automatically handles all the events and data | ||
| await pipeline( | ||
| clientResponse, | ||
| response, | ||
| ); | ||
|
|
||
| resolve(); | ||
| } catch (error) { | ||
| // Client error, pipeline already destroys the streams, ignore. | ||
| resolve(); | ||
| } | ||
| }); | ||
|
|
||
| client.once('socket', (socket) => { | ||
| countTargetBytes(request.socket, socket); | ||
| }); | ||
|
|
||
| // Can't use pipeline here as it automatically destroys the streams | ||
| request.pipe(client); | ||
| client.on('error', (error: NodeJS.ErrnoException) => { | ||
| if (response.headersSent) { | ||
| return; | ||
| } | ||
|
|
||
| const statusCode = errorCodeToStatusCode[error.code!] ?? badGatewayStatusCodes.GENERIC_ERROR; | ||
|
|
||
| response.statusCode = statusCode; | ||
| response.setHeader('content-type', 'text/plain; charset=utf-8'); | ||
| response.end(http.STATUS_CODES[response.statusCode]); | ||
|
|
||
| resolve(); | ||
| }); | ||
| }); |
Oops, something went wrong.