NodeJS blocks response from CONNECT

[Jiri-Mihal]

On my Linode VPS (Nanode) I host two separated apps: file streamer and forward proxy. The file streamer is written in NodeJS and streams files from remote servers to clients. A forward proxy is a classic forward proxy. The problem I have is that after a while of serving files, file streamer starts to block responses from the proxy. Connections to the proxy are ok, also file streamer responds normally.

As a proxy, I've tried Squid, Tinyproxy or even simple NodeJS proxy. All result to the same error, broken pipe. It seems that file streamer somehow blocks responses from different apps/streams. Even if the file streamer no longer sends any files, it still blocks the proxy. Once this condition happens, the only thing it helps is restarting the file streamer.

Here is the simplified code of file streamer. It uses pipes, there is no magic:

// File Streamer
const server = http.createServer((sreq, sres) => {
    const url = 'some user defined url';
    const options = {some options};
    const req = https.request(url, options, (res) => {
        sres.writeHead(res.statusCode, res.headers);
        res.pipe(sres);
    });
    req.end();
});

Do you have some ideas about why the file streamer (NodeJS) blocks responses from the other apps? I didn't detect any memory leaks, any CPU overloading, any high I/O operations, etc.

UPDATE:
I tried to merge File Streamer script (above) and Forward Proxy script into one server. Unfortunately, it didn't help. Interesting is, that File Streamer works properly, but Forward Proxy returns Error: write EPIPE.

// Forward Proxy
server.on('connect', (req, socket, head) => {
    debug('CONNECT');
    const srvUrl = url.parse(`http://${req.url}`);
    const srvSocket = net.connect(srvUrl.port, srvUrl.hostname, () => {
        socket.write('HTTP/1.1 200 Connection Established\r\n' +
            '\r\n');
        srvSocket.write(head);
        srvSocket.pipe(socket);
        socket.pipe(srvSocket);
        socket.on('error', (err) => {
            debug('Some socket error.');
            debug(err.stack); // Error: write EPIPE <--- ERROR
        });
    });
});

UPDATE 2:
I believe that problem must be in the file streamer part. But I wasn't able to determine where. Probably it's something wrong with NodeJS on low level. Now I stream files only with NGiNX, it's flawless and it consumes 3x less memory.

[ronag]

@Jiri-Mihal: Do you perhaps have a full repo I can use to try and see where it goes wrong for you? i.e. with the node based proxy.

[Jiri-Mihal]

@ronag thank you. Below is the full code. The only thing I changed are pairs in decryptStrtr().

package.json

{
  "name": "public",
  "version": "1.0.0",
  "dependencies": {
    "ebg13": "^1.3.9"
  }
}

app.js

const http = require('http');
const https = require('https');
const crypto = require('crypto');
const ebg13 = require('ebg13');

// Show debug messages?
const showDebugMessages = typeof process.argv[2] !== 'undefined';

function debug(message) {
    if (showDebugMessages) {
        console.log(message);
    }
}

// Show number of connections
function getNumOfConnections() {
    server.getConnections((error, count) => {
        debug(count);
    });
}

if (showDebugMessages) {
    setInterval(() => {
        getNumOfConnections();
    }, 1000);
}

function initialOptionsPromise(headers) {
    return new Promise((resolve, reject) => {
        try {
            delete headers.host;
            resolve({
                method: 'GET',
                headers: headers
            });
        } catch (e) {
            reject();
        }
    });
}

function redirPromise(maxRedirs) {
    return new Promise((resolve, reject) => {
        if (maxRedirs > 0) {
            maxRedirs--;
            resolve(maxRedirs);
        } else {
            reject();
        }
    });
}

function cookiesPromise(headers) {
    return new Promise((resolve, reject) => {
        let cookie = '';
        if (headers.hasOwnProperty('set-cookie')) {
            headers['set-cookie'].forEach((value) => {
                cookie += value.substr(0, value.indexOf(';')) + '; ';
            });
        }
        if (cookie) {
            resolve(cookie);
        } else {
            reject();
        }
    });
}

function manageRequest(url, options, req, res, sres, maxRedirs, fileName, fileExt) {
    if (res.headers.hasOwnProperty('location')) {
        req.abort();
        redirPromise(maxRedirs)
            .then((maxRedirs) => {
                cookiesPromise(res.headers)
                    .then((cookies) => {
                        options.headers['cookie'] = cookies;
                        httpRequest(url, options, sres, fileName, fileExt, maxRedirs);
                    }, () => {
                        httpRequest(url, options, sres, fileName, fileExt, maxRedirs);
                    });
            }, () => {
                // Too many redirection
                debug('Too many redirection');
                sres.end();
            });
    } else {
        filenamePromise(fileName, fileExt, res.headers)
            .then((fileName) => {
                sres.setHeader('Content-Disposition', 'attachment; filename="' + fileName + '"');
                sres.writeHead(res.statusCode, res.headers);
                res.pipe(sres);
            }, () => {
                debug('Invalid fileName: ' + fileName + ', fileExt: ' + fileExt + ', url: ' + url);
                req.abort();
                sres.end();
            });
    }
}

function httpRequest(url, options, sres, fileName, fileExt, maxRedirs = 4) {
    if (url.match(/^https:/)) {
        const req = https.request(url, options, (res) => {
            manageRequest(url, options, req, res, sres, maxRedirs, fileName, fileExt);
        });
        req.end();
    } else if (url.match(/^http:/)) {
        const req = http.request(url, options, (res) => {
            manageRequest(url, options, req, res, sres, maxRedirs, fileName, fileExt);
        });
        req.end();
    } else {
        // Invalid URL
        sres.end();
    }
}

function getExtFromMime(mime) {
    let ext = false;
    const mimes = {
        'application/xml': 'xml',
        'application/vnd.apple.mpegurl': 'm3u8',
        'audio/mp3': 'mp3',
        'audio/mp4': 'm4a',
        'audio/mpeg': 'mpga',
        'audio/ogg': 'oga',
        'audio/webm': 'weba',
        'audio/x-aac': 'aac',
        'audio/x-aiff': 'aif',
        'audio/x-flac': 'flac',
        'audio/x-mpegurl': 'm3u',
        'audio/x-ms-wma': 'wma',
        'audio/x-wav': 'wav',
        'image/bmp': 'bmp',
        'image/gif': 'gif',
        'image/jpeg': 'jpg',
        'image/png': 'png',
        'image/svg+xml': 'svg',
        'image/tiff': 'tiff',
        'text/vnd.dvb.subtitle': 'sub',
        'image/webp': 'webp',
        'video/3gpp': '3gp',
        'video/3gpp2': '3g2',
        'video/h261': 'h261',
        'video/h263': 'h263',
        'video/h264': 'h264',
        'video/mp4': 'mp4',
        'video/mpeg': 'mpeg',
        'video/ogg': 'ogv',
        'video/webm': 'webm',
        'video/x-f4v': 'f4v',
        'video/x-fli': 'fli',
        'video/x-flv': 'flv',
        'video/x-m4v': 'm4v',
        'video/x-matroska': 'mkv',
        'video/x-ms-wm': 'wm',
        'video/x-ms-wmv': 'wmv',
        'video/x-msvideo': 'avi'
    };
    if (mimes.hasOwnProperty(mime)) {
        ext = mimes[mime];
    } else {
        debug('Unsupported mime extenstion: ' + mime);
    }
    return ext;
}

function filenamePromise(name, ext, headers) {
    return new Promise((resolve, reject) => {
        const fileName = typeof name === 'string' && name ? name : 'download';
        let mime = false;

        if (headers.hasOwnProperty('content-type')) {
            mime = headers["content-type"];
        }
        if (headers.hasOwnProperty('Content-Type')) {
            mime = headers["Content-Type"];
        }

        const fileExtension = mime ? getExtFromMime(mime) : ext;
        if (fileExtension) {
            resolve(fileName + '.' + fileExtension);
        } else {
            reject();
        }
    });
}

function decryptStrtr(text) {
    const pairs = {
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a",
        "a": "a"
    };

    let decryptedText = '';
    for (let i = 0; i < text.length; i++) {
        if (pairs.hasOwnProperty(text[i])) {
            decryptedText += pairs[text[i]];
        } else {
            decryptedText += text[i];
        }
    }
    return decryptedText;
}

function fastDecrypt(string) {

    string = decodeURIComponent(string);

    // Get control hash
    const controlHash = string.slice(-32);
    if (!controlHash) {
        return false;
    }

    // Get encrypted data
    string = string.slice(0, -32);

    // Compare control hash with hash of encrypted data
    const hash = crypto.createHash('md5');
    hash.update(string);
    const hashedString = hash.digest('hex');
    if (controlHash !== hashedString) {
        return false;
    }

    // Decode string
    string = ebg13(string);
    string = Buffer.from(string, 'base64').toString('ascii');
    string = decryptStrtr(string);
    return string;
}

function dataPromise(url) {
    return new Promise((resolve, reject) => {
        try {
            if (url.length >= 3000) {
                // Don't allow longer URLs to prevent attacks blocking memory and cpu
                reject();
            } else {
                if (url[0] === '/') {
                    url = url.substr(1);
                }
                let data = fastDecrypt(url);
                data = JSON.parse(data);
                resolve(data);
            }
        } catch (e) {
            reject();
        }
    });
}

// File Streamer
const server = http.createServer((sreq, sres) => {
    dataPromise(sreq.url)
        .then((data) => {
            // Check all required properties and try download file
            if (
                typeof data === 'object'
                && data.hasOwnProperty('ip')
                && data.ip === sreq.headers['x-forwarded-for']
                && data.hasOwnProperty('ex')
                && Math.round(Date.now() / 1000) <= data.ex
                && data.hasOwnProperty('lu')
                && data.hasOwnProperty('lt')
                && data.hasOwnProperty('fn')
                && data.hasOwnProperty('fe')
                && data.lt === 2
                && !sreq.url.match(/\/favicon.ico/)
            ) {
                // Use server's request headers for HTTP(s) request,
                // but remove the host header (we can't set it)
                const options = initialOptionsPromise(sreq.headers);
                options
                    .then((options) => {
                        try {
                            const url = data.lu.replace(/&\//g, '/');
                            httpRequest(url, options, sres, data.fn, data.fe);
                        } catch (e) {
                            debug(e.message);
                            sres.statusCode = 404;
                            sres.end();
                        }
                    }, () => {
                        // Can't prepare initial options
                        debug('Can\'t prepare initial options');
                        sres.statusCode = 404;
                        sres.end();
                    });
            } else {
                // Data don't match required properties
                let dataError = 'Error(s): ';
                if (typeof data !== 'object') {
                    dataError += 'not an object, ';
                }
                if (!data.hasOwnProperty('ip')) {
                    dataError += 'ip, ';
                }
                if (data.ip !== sreq.headers['x-forwarded-for']) {
                    dataError += data.ip + '!==' + sreq.headers['x-forwarded-for'] + ', ';
                }
                if (!data.hasOwnProperty('ex')) {
                    dataError += 'ex, ';
                }
                if (Math.round(Date.now() / 1000) > data.ex) {
                    dataError += 'expired, ';
                }
                if (!data.hasOwnProperty('lu')) {
                    dataError += 'lu, ';
                }
                if (!data.hasOwnProperty('lt')) {
                    dataError += 'lt, ';
                }
                if (!data.hasOwnProperty('fn')) {
                    dataError += 'fn, ';
                }
                if (!data.hasOwnProperty('fe')) {
                    dataError += 'fe, ';
                }
                if (data.lt !== 2) {
                    dataError += 'lt === ' + data.lt + ', ';
                }
                if (sreq.url.match(/\/favicon.ico/)) {
                    dataError += 'favicon';
                }

                debug('Data don\'t match required properties. ' + dataError);
                sres.statusCode = 404;
                sres.end();
            }

        }, () => {
            // Invalid data
            debug('Invalid data');
            sres.statusCode = 404;
            sres.end();
        });
}).listen(8080);

// Forward Proxy
server.on('connect', (req, socket, head) => {
    debug('CONNECT');
    const srvUrl = url.parse(`http://${req.url}`);
    const srvSocket = net.connect(srvUrl.port, srvUrl.hostname, () => {
        socket.write('HTTP/1.1 200 Connection Established\r\n' +
            '\r\n');
        srvSocket.write(head);
        srvSocket.pipe(socket);
        socket.pipe(srvSocket);
        socket.on('error', (err) => {
            debug('Some socket error.');
            debug(err.stack);
        });
    });
});

[ronag]

@Jiri-Mihal: Do you think you could remove the stuff that are not relevant for reproducing the issue?

[Jiri-Mihal]

@ronag below you can find reduced code. I added some comments to make code more clear.

const http = require('http');
const https = require('https');
const net = require('net');
const url = require('url');

// Show debug messages?
const showDebugMessages = typeof process.argv[2] !== 'undefined';
function debug(message) {
    if (showDebugMessages) {
        console.log(message);
    }
}

// Prepare http request options from server request
function initialOptionsPromise(headers) {
    return new Promise((resolve, reject) => {
        try {
            // Use server's request headers for HTTP(s) request,
            // but remove the host header (we can't set it)
            delete headers.host;
            resolve({
                method: 'GET',
                headers: headers
            });
        } catch (e) {
            reject();
        }
    });
}

// Limit number of redirects
function redirPromise(maxRedirs) {
    return new Promise((resolve, reject) => {
        if (maxRedirs > 0) {
            maxRedirs--;
            resolve(maxRedirs);
        } else {
            reject();
        }
    });
}

// Convert values from set-cookie headers to cookie header
function cookiesPromise(headers) {
    return new Promise((resolve, reject) => {
        let cookie = '';
        if (headers.hasOwnProperty('set-cookie')) {
            headers['set-cookie'].forEach((value) => {
                cookie += value.substr(0, value.indexOf(';')) + '; ';
            });
        }
        if (cookie) {
            resolve(cookie);
        } else {
            reject();
        }
    });
}

// Manage http(s) request
function manageRequest(url, options, req, res, sres, maxRedirs) {
    // Manage redirection
    if (res.headers.hasOwnProperty('location')) {
        req.abort();
        redirPromise(maxRedirs)
            .then((maxRedirs) => {
                cookiesPromise(res.headers)
                    .then((cookies) => {
                        options.headers['cookie'] = cookies;
                        httpRequest(url, options, sres, maxRedirs);
                    }, () => {
                        httpRequest(url, options, sres, maxRedirs);
                    });
            }, () => {
                // Too many redirection
                debug('Too many redirection');
                sres.end();
            });
    } else {
    // Return final response
        sres.writeHead(res.statusCode, res.headers);
        res.pipe(sres);
    }
}

// Create http(s) request
function httpRequest(url, options, sres, maxRedirs = 4) {
    if (url.match(/^https:/)) {
        const req = https.request(url, options, (res) => {
            manageRequest(url, options, req, res, sres, maxRedirs);
        });
        req.end();
    } else if (url.match(/^http:/)) {
        const req = http.request(url, options, (res) => {
            manageRequest(url, options, req, res, sres, maxRedirs);
        });
        req.end();
    } else {
        // Invalid URL
        sres.end();
    }
}

// File Streamer
const server = http.createServer((sreq, sres) => {
    
    const options = initialOptionsPromise(sreq.headers);
    options
        .then((options) => {
            try {
                const url = data.lu.replace(/&\//g, '/');
                httpRequest(url, options, sres);
            } catch (e) {
                debug(e.message);
                sres.statusCode = 404;
                sres.end();
            }
        }, () => {
            // Can't prepare initial options
            debug('Can\'t prepare initial options');
            sres.statusCode = 404;
            sres.end();
        });


}).listen(8080);

// Forward Proxy
server.on('connect', (req, socket, head) => {
    debug('CONNECT');
    const srvUrl = url.parse(`http://${req.url}`);
    const srvSocket = net.connect(srvUrl.port, srvUrl.hostname, () => {
        socket.write('HTTP/1.1 200 Connection Established\r\n' +
            '\r\n');
        srvSocket.write(head);
        srvSocket.pipe(socket);
        socket.pipe(srvSocket);
        socket.on('error', (err) => {
            debug('Some socket error.');
            debug(err.stack);
        });
    });
});

[ronag]

@Jiri-Mihal: That's one step in the right direction.

• How do I reproduce the problem given that code? It doesn't seem to include any actual request or logging to provoke or show any problem? i.e. running it doesn't actually do anything?
• Is https relevant for reproducing?
• Are cookie handling relevant for reproducing?
• Is redirection relevant for reproducing?

Please keep it as minimal as possible to just show the problem.

[Jiri-Mihal]

@ronag my answers will be not so clear...

• users stream files from many sources, so it's hard to give you one specific source. For testing purpose, you can try these sources https://www.linode.com/speedtest
• most of the requests are to https addresses
• a minority of requests include cookies or location header

I'm not able to reproduce the error on my local server (different OS, different network properties). The weirdest thing is that once NodeJS on production starts to block responses from 'CONNECT', then it keeps it blocking even there is no load. And it blocks them system-wide, not just inside the NodeJS (as I described above).