程序浅谈 工具
2022-07-29
websocket本质是一种网络应用层协议，建立在单个TCP连接上的全双工模式，用来弥补了http协议在持续双向通信能力上的不足，允许服务端与客户端之间可以双向主动推送数据。
特点：
ws://${domain}:80/${path}，或者wss://${domain}:443/${path}采用ws库快速构建一个websocket server，监听connection事件，收到消息并且打印后，立马发送给客户端
const ws = require('ws');
let wsServer = new ws.Server({
port: 3000,
host:'127.0.0.1',
path:'/websocket'
});
wsServer.on('connection', function (server) {
console.log('client connected');
server.on('message', function (message) {
console.dir(message)
console.log(message.toString());
server.send(`hello:${message}`)
});
});
复制代码快速搭建一个websocket client，利用http-server在目录下启动，并且访问该页面
<!DOCTYPE html>
<html>
<head>
<title>websocket demo</title>
</head>
<body>
<h1></h1>
<br>
<input type='text' id='sendText'>
<button onclick='send()'>send</button>
</body>
</html>
<script>
const ws = new WebSocket('ws://127.0.0.1:3000/websocket');
ws.onopen = function () {
console.log('服务器连接')
}
ws.onmessage = (msg) => {
console.log('来自服务器发来的数据：', msg)
alert('服务器返回内容：' + msg.data)
}
ws.onclose = () => {
console.log('服务器关闭')
}
function send() {
if (ws) {
let msg = document.getElementById('sendText').value;
ws.send(msg)
} else {
alert('websocket server error')
}
}
</script>
复制代码先启动websocket server，然后浏览器请求websocket client页面，抓包请求如下：
前三条为tcp的三次握手信息，既然谈到了，为了文章的完整性，还是简单描述一下；
三次握手必要性：
SYN（连接请求）的攻击危害：
攻击方发送海量伪造源IP的第一次握手SYN包，将服务器的半连接队列给打满（超过最大值），正常的客户发送SYN数据包请求连接就会被服务器丢弃，导致正常的连接请求无法成功，严重引起网络堵塞甚至系统瘫痪
规避方式：
限制ip连接次数（限制同一IP一分钟内新建立的连接数仅为10）；增大半连接状态的连接数容量（会增大内存资源占用，/etc/sysctl.d/sysctl.conf ，字段tcp_max_syn_backlog）
server的接收窗口大小发生了变化，可以正常接收数据了，就会出现这一条记录
抓包分析看出，websocket通信在双方TCP三次握手成功后，还需要发送一次额外的http请求，才能正式建立连接。http请求报文如下：
GET /websocket HTTP/1.1
Host: 127.0.0.1:3000
Connection: Upgrade
Pragma: no-cache
Cache-Control: no-cache
User-Agent: Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/103.0.0.0 Safari/537.36
Upgrade: websocket
Origin: http://127.0.0.1:8080
Sec-WebSocket-Version: 13
Accept-Encoding: gzip, deflate, br
Accept-Language: en-US,en;q=0.9,zh-CN;q=0.8,zh;q=0.7
Sec-WebSocket-Key: Ap4ZCLgwbnDQ2ump+7ea3g==
Sec-WebSocket-Extensions: permessage-deflate; client_max_window_bits
HTTP/1.1 101 Switching Protocols
Upgrade: websocket
Connection: Upgrade
Sec-WebSocket-Accept: Ih1TB0gxAY3zGzvQCYrIeM5bEdw=
复制代码请求headers的限定：
响应header分析：
只有状态码为101，才表示服务端同意了协议升级，对于其他状态码，client会根据语义相应处理；
client会检测响应headers中是否包含Upgrade字段，且检测值是否为websokcet（不区分大小写），若缺失或不匹配，会自动终止连接；
client会检测响应headers中是否包含Sec-WebSocket-Protocol字段，并校验它的合理性，若缺失或校验失败，会在自动终止连接；
Sec-WebSocket-Protocol校验算法（client & server的约定）：server收到Sec-WebSocket-Key后，会将其与websocket魔数258EAFA5-E914-47DA- 95CA-C5AB0DC85B11进行字符串拼接，即${Sec-WebSocket-Key}258EAFA5-E914-47DA- 95CA-C5AB0DC85B11，然后对它做SHA1哈希运算后再做一次base64编码，就为Sec-WebSocket-Protocol。
握手通过后，双方就是长连接了，可以随时进行双向数据的传输。
由上文可知，除去tcp三次握手外，websocket真实的建立连接是那次关键的http请求，那其实可以针对它来做一层http网关来代理后续的数据传输了。
先描述config.json文件：
json格式，websocketTestOne key代表一个webSocket，根下文协议升级请求的path相呼应，即一个该配置对应的代理请求地址应该为：http://{domain}/websocketTestOne，添加多个配置，依次类推
{
"websocketTestOne": {
"host": "127.0.0.1",
"port": "3000"
}
}
复制代码httpServer.js，如下所示，代码量不多，简单介绍一下流程：
upgrade，而不是request，upgrade事件中，针对clientSocket一系列监听的预处理；request.url解析出来的path在config中找不到，结束clientSocket；data事件，将报文写入serverSocket，serverSocket监听data事件，将报文写入clientSocket，交替进行；data监听事件；/**
* create by ikejcwang on 2022.07.25.
* 注：这只是一个测试的demo
*/
'use strict';
const http = require('http');
const nodeUtil = require('util');
const URL = require('url');
const net = require('net');
const settings = require('./settings').settings;
const configs = require('./settings').configs;
const connectTimeout = settings['connectTimeout'] ? settings['connectTimeout'] : 5000; // 建立连接的超时设定
const connectKeepalive = settings['connectKeepalive'] ? settings['connectKeepalive'] : 60000; // 连接后的keepalive超时设定
const socketTimeout = settings['socketTimeout'] ? settings['socketTimeout'] : 60000; // socket的timeout，
httpServer();
/**
* 启动入口
*/
function httpServer() {
console.dir(settings)
startHttpServer();
}
function startHttpServer() {
let server = http.createServer();
server.on('upgrade', listenUpgradeEvent);
server.on('request', listenRequestEvent);
server.on('close', () => {
console.log('http Server has Stopped At:' + settings['bindPort'])
});
server.on('error', err => {
console.log('http Server error:' + err.toString());
setTimeout(() => {
process.exit(1);
}, 3000);
});
server.listen(settings['bindPort'], settings['bindIP'], settings['backlog'] || 8191, () => {
console.log('Started Http Server At: ' + settings['bindIP'] + ':' + settings['bindPort']);
});
}
/**
* 监听upgrade事件
* @param request
* @param cliSocket
* @param header
* @returns {Promise<void>}
*/
async function listenUpgradeEvent(request, cliSocket, header) {
let serverSocket = null;
cliSocket.on('error', e => {
if (serverSocket) {
serverSocket.destroy();
}
logInfo('cliSocket has error', nodeUtil.inspect(e))
});
cliSocket.on('end', () => {
logInfo('cliSocket has ended');
});
cliSocket.on('close', function () {
logInfo('cliSocket has closed');
});
cliSocket.setTimeout(socketTimeout, () => {
cliSocket.destroy(new Error('timeout'));
if (serverSocket) {
serverSocket.destroy();
}
})
try {
if (!configs || Object.keys(configs).length < 1) {
cliSocket.end();
return;
}
let sourceUrl = URL.parse(request.url, true);
let pathArr = sourceUrl.pathname.split('/');
if (pathArr.length === 1) {
cliSocket.end();
return;
}
let websocketName = pathArr[1];
if (!websocketName || !configs[websocketName]) {
cliSocket.end();
return;
}
serverSocket = await connectSocket(configs[websocketName]);
serverSocket.on('error', err => {
cliSocket.end();
logInfo('server socket error', nodeUtil.inspect(err));
});
cliSocket.on('data', chunk => {
cliSocket.pause(); // 收到数据后，暂停当前cliSocket
if (serverSocket.write(chunk)) {
cliSocket.resume(); // server socket写成功后，在激活当前cliSocket
}
}).on('end', () => {
console.log('end')
serverSocket.end(); // 双写完处理
});
serverSocket.on('data', chunk => {
serverSocket.pause();
if (cliSocket.write(chunk)) {
serverSocket.resume();
} else {
cliSocket.once('drain', () => serverSocket.resume()); // 如果调用 stream.write(chunk) 返回 false，则当可以继续写入数据到流时会触发 drain 事件
}
}).on('end', () => {
cliSocket.end()
});
let connectHeaders = request.headers;
connectHeaders['host'] = `${configs[websocketName].host}:${configs[websocketName].port}`;
let headersTemp = '';
for (let key in connectHeaders) {
headersTemp += `${key}: ${connectHeaders[key]}\r\n`
}
serverSocket.write(`${request.method} ${request.url} HTTP/1.1\r\n${headersTemp}\r\n`); // 向真实的webSocket服务开始握手连接
if (header && header.length > 0) {
serverSocket.write(header)
}
} catch (e) {
if (cliSocket.writable) {
cliSocket.write(`HTTP/1.1 502 Server UnReachable\r\n\r\n`);
}
cliSocket.end();
console.log(`request_error: ${nodeUtil.inspect(e)}`);
}
}
/**
* 监听request事件
* @param request
* @param response
* @returns {Promise<void>}
*/
async function listenRequestEvent(request, response) {
// 再次证实websocket握手时到不了这里，因为headers信息的connection字段为Upgrade，触发的是Upgrade事件
console.log('listenRequestEvent')
}
/**
* 连接socket
* @param websocketConfig
* @returns {Promise<unknown>}
*/
function connectSocket(websocketConfig) {
return new Promise((resolve, reject) => {
let socket = net.connect(websocketConfig);
let timer = setTimeout(() => {
socket.removeListener('error', onError)
socket.destroy();
reject(Object.assign(new Error('connect timeout'), websocketConfig))
}, connectTimeout);
let onConnect = () => {
socket.setKeepAlive(true, connectKeepalive);
socket.removeListener('error', onError)
clearInterval(timer);
// TODO 创建tcp连接时，默认都会启用Nagle算法，此处禁用它，(Nagle试图以延迟为代价来优化吞吐量，但是我们并不需要)，传参true或不传即禁用，
socket.setNoDelay();
socket.setTimeout(socketTimeout + 60000, () => {
socket.destroy(new Error('socket server timeout'));
})
resolve(socket);
}
let onError = e => {
clearInterval(timer);
reject(e);
}
socket.once('connect', onConnect);
socket.once('error', onError);
});
}
function logInfo(...args) {
console.dir(args)
}
复制代码webSocketServer.js，比较简单，使用ws模块快速构建；
连接建立，输出信息，收到报文，输出报文，并添加前缀原路发出去；
const ws = require('ws');
let wsServer = new ws.Server({
port: 3000,
host:'127.0.0.1',
});
wsServer.on('connection', function (server) {
console.log('client connected');
server.on('message', function (message) {
console.dir(message)
console.log(message.toString());
server.send(`hello:${message}`)
});
});
