ntrip-caster服务后端

2026-05-12 17:43:03 +08:00
commit 6e761f3727
45 changed files with 3125 additions and 0 deletions
--- a/app/middleware/ntripCaster/ntripCaster.js
+++ b/app/middleware/ntripCaster/ntripCaster.js
@@ -0,0 +1,364 @@
+const net = require('net');
+const CryptoJS = require("crypto-js");
+
+const KEY = CryptoJS.enc.Utf8.parse('BDQH');
+
+class SimpleNtripCaster {
+    constructor(app) {
+        this.app = app;
+        const { config, redis } = app;
+        const port = config.caster.port;
+        this.port = port;
+        this.redis = redis;
+        this.sourcetable = [];
+        this.serverSockets = new Map(); // mountpoint -> server socket
+        this.sizeSockets = new Map(); // client socket -> mountpoint
+        this.clientSockets = new Map(); // client socket -> mountpoint
+        this.mountData = new Map(); // mountpoint -> 当前连接的 server socket
+        console.log(`[Caster] 启动在端口 ${port}`);
+    }
+
+    start() {
+        this.server = net.createServer((socket) => {
+            let clientInfo = '';
+            socket.on('data', (data) => {
+                const request = data.toString();
+                clientInfo += request;
+                // 检查是否收到完整的HTTP请求（以空行结束）
+                if (clientInfo.includes('\r\n\r\n')) {
+                    this.handleRequest(socket, clientInfo);
+                    clientInfo = '';
+                }
+            });
+            socket.on('end', () => {
+                console.log('[Caster] 连接关闭');
+                this.removeClient(socket);
+                this.broadSizeData();
+            });
+            socket.on('error', (err) => {
+                console.log('[Caster] 连接错误:', err.message);
+                this.removeClient(socket);
+                this.broadSizeData();
+            });
+        });
+        this.server.listen(this.port, () => {
+            console.log(`[Caster] 等待连接...`);
+        });
+    }
+
+    sendSourcetable(socket, statusCode, statusMessage, headers1 = {}, headers2 = {}, body = '') {
+        let response = `HTTP/1.1 ${statusCode} ${statusMessage}\r\n`;
+        for (const [key, value] of Object.entries(headers1)) {
+            response += `${key}: ${value}\r\n`;
+        }
+        response += `\r\n`;
+        response += `SOURCETABLE ${statusCode} ${statusMessage}\r\n`;
+        for (const [key, value] of Object.entries(headers2)) {
+            response += `${key}: ${value}\r\n`;
+        }
+        response += '\r\n' + body;
+        socket.end(response);
+    }
+
+    sendResponse(socket, statusCode, statusMessage, headers = {}, body = '') {
+        let response = `HTTP/1.1 ${statusCode} ${statusMessage}\r\n`;
+        for (const [key, value] of Object.entries(headers)) {
+            response += `${key}: ${value}\r\n`;
+        }
+        response += '\r\n' + body;
+        socket.end(response);
+    }
+
+    clientSendResponse(socket, statusCode, statusMessage, headers = {}, body = '') {
+        let response = `HTTP/1.1 ${statusCode} ${statusMessage}\r\n`;
+        for (const [key, value] of Object.entries(headers)) {
+            response += `${key}: ${value}\r\n`;
+        }
+        response += '\r\n' + body;
+        socket.write(response);
+    }
+
+    handleOptions(socket, headers) {
+        this.sendResponse(socket, 204, 'No Content', {
+            'Access-Control-Allow-Origin': headers.origin,
+            'Access-Control-Allow-Methods': 'GET, OPTIONS, SIZE, SOURCE',
+            'Access-Control-Allow-Headers': 'Content-Type, uid, Authorization, ntrip-version, Client-Type, Request-Type',
+        });
+    }
+
+    parseHeaders(headerLines) {
+        const headers = {};
+        headerLines.forEach((line) => {
+            const [key, value] = line.split(': ');
+            headers[key.toLowerCase()] = value;
+        });
+        return headers;
+    }
+
+    handleRequest(socket, request) {
+        const lines = request.split('\r\n');
+        const firstLine = lines[0];
+        console.log(`[Caster] 收到请求: ${firstLine.substring(0, 50)}...`);
+        // 处理 OPTIONS 请求
+        if (firstLine.split(' ')[0] === 'OPTIONS') {
+            const headers = this.parseHeaders(lines.slice(1, lines.indexOf('')));
+            this.handleOptions(socket, headers);
+        } else if (firstLine === 'GET / HTTP/1.1' || firstLine === 'GET / HTTP/1.0' || firstLine === 'GET /') {
+            // 客户端请求源列表
+            this.handleSourcetable(socket, lines);
+        } else if (firstLine.startsWith('SOURCE')) {
+            // Server 挂载请求
+            this.handleServerMount(socket, lines);
+        } else if (firstLine.startsWith('GET /')) {
+            // 客户端请求数据流
+            this.handleClientRequest(socket, lines);
+        } else {
+            this.sendResponse(socket, 400, 'Bad Request', {
+                'Access-Control-Allow-Origin': '*'
+            });
+        }
+    }
+
+    // 发送源列表
+    handleSourcetable(socket, lines) {
+        const headers = this.parseHeaders(lines.slice(1, lines.indexOf('')));
+        let table = '';
+        this.sourcetable.forEach(element => {
+            const { mountpoint, MountpointOptions } = element;
+            // 源ID
+            const SourceID = MountpointOptions[0] || 'STR: ';
+            // 格式
+            const Format = MountpointOptions[1] || 'RTCM';
+            // 格式详情
+            const FormatDetails = MountpointOptions[2] || '';
+            // 载体
+            const Carrier = MountpointOptions[3] || '';
+            // 导航系统
+            const NavSystem = MountpointOptions[4] || '';
+            // 网络
+            const NetWork = MountpointOptions[5] || '';
+            // 国家
+            const Country = MountpointOptions[6] || '';
+            // 纬度
+            const Latitude = MountpointOptions[7] || 0.0;
+            // 经度
+            const Longitude = MountpointOptions[8] || 0.0;
+            // NMEA
+            const NMEA = MountpointOptions[9] || '';
+            // 解决方案
+            const Soution = MountpointOptions[10] || '';
+            // 发生器
+            const Generator = MountpointOptions[11] || '';
+            // 压缩加密
+            const ComprEncryp = MountpointOptions[12] || '';
+            // 认证
+            const Auth = MountpointOptions[13] || '';
+            // 费用
+            const Fee = MountpointOptions[14] || '';
+            // 比特率
+            const Bitrate = MountpointOptions[15] || '';
+            const found = lines.find(line => line.startsWith('Ntrip-Version: Ntrip/2.0'));
+            if (found == 'Ntrip-Version: Ntrip/2.0') {
+                // 变体B：RTCM/ntrip-2风格
+                table += `STR;${Format};${mountpoint};${FormatDetails};;${Carrier};${NavSystem};${NetWork};${Country};${Latitude};${Longitude};${NMEA};${SourceID};;;;\r\n`;
+            } else {
+                // 变体A：BKG/NTRIP 1.0风格
+                table += `STR;${Format};${mountpoint};${FormatDetails};;${Carrier};${NavSystem};${SourceID};${Latitude};${Longitude};0.0;${NetWork};${Country};;;;\r\n`;
+            }
+        });
+        table += 'ENDSOURCETABLE\r\n';
+        this.sendSourcetable(socket, 200, 'OK', {
+            'Access-Control-Allow-Origin': headers.origin ? headers.origin : '*',
+            'Server': 'NTRIP Caster',
+            'Content-Type': 'text/plain',
+            'Date': 'Tue, 12 May 2026 08:00:00 GMT',
+            'Connection': 'close'
+        }, {
+            'Access-Control-Allow-Origin': headers.origin ? headers.origin : '*',
+            'Server': 'Caster/1.0',
+            'Content-Type': 'text/plain',
+            'Charset': 'UTF-8'
+        }, table);
+        console.log('[Caster] 已发送源列表');
+    }
+
+    // 处理 Server 挂载
+    async handleServerMount(socket, lines) {
+        const source = lines[0].split(' ');
+        // 密码
+        const password = source[1];
+        // 挂载点
+        const mountpoint = source[2];
+        const findMountpoint = await this.app.curl('http://127.0.0.1:' + this.app.server.address().port + '/findMountpoint', {
+            dataType: 'json',
+            contentType: 'application/x-www-form-urlencoded',
+            method: 'POST',
+            timeout: 3000,
+            data: {
+                mountpoint: mountpoint
+            }
+        });
+        if (findMountpoint.data.code == 404) {
+            this.sendResponse(socket, 404, 'Mountpoint Not Found', {});
+            return;
+        }
+        if (password !== findMountpoint.data.password) {
+            this.sendResponse(socket, 401, 'Unauthorized', {});
+            return;
+        }
+        const MountpointOptions = lines[2].split(';');
+        this.sourcetable.push({ password, mountpoint, MountpointOptions });
+        this.serverSockets.set(mountpoint, socket);
+        this.mountData.set(mountpoint, {
+            socket: socket,
+            lastUpdate: Date.now()
+        });
+        // 响应 Server
+        socket.write('ICY 200 OK\r\n\r\n');
+        console.log(`[Caster] Server 挂载成功: ${mountpoint}`);
+        // 监听 Server 数据
+        socket.on('data', (data) => {
+            this.broadcastData(mountpoint, data);
+        });
+        socket.on('end', () => {
+            console.log(`[Caster] Server 断开: ${mountpoint}`);
+            this.serverSockets.delete(mountpoint);
+            this.mountData.delete(mountpoint);
+        });
+    }
+
+    // 处理 Client 请求
+    async handleClientRequest(socket, lines) {
+        // 解析请求的挂载点
+        const headers = this.parseHeaders(lines.slice(1, lines.indexOf('')));
+        const firstLine = lines[0];
+        const match = firstLine.match(/GET \/(\w+) HTTP\/1\.[01]/);
+        if (!match) {
+            this.sendResponse(socket, 404, '404 Not Found', {
+                'Access-Control-Allow-Origin': headers.origin,
+            });
+            return;
+        }
+        const mountpoint = match[1];
+        console.log(`[Caster] Client 请求挂载点: ${mountpoint}`);
+        const find = lines.find(line => line.startsWith('Request-Type: SIZE'));
+        const found = lines.find(line => line.startsWith('Client-Type: Virtual-NTRIP-Client/1.0'));
+        if (found !== 'Client-Type: Virtual-NTRIP-Client/1.0') {
+            if (find !== 'Request-Type: SIZE') {
+                const findMountpoint = await this.app.curl('http://127.0.0.1:' + this.app.server.address().port + '/findMountpoint', {
+                    dataType: 'json',
+                    contentType: 'application/x-www-form-urlencoded',
+                    method: 'POST',
+                    timeout: 3000,
+                    data: {
+                        mountpoint: mountpoint
+                    }
+                });
+                if (findMountpoint.data.code == 404) {
+                    this.sendResponse(socket, 404, 'Mountpoint Not Found', {});
+                    return;
+                }
+                const getPassword = await this.app.curl('http://127.0.0.1:' + this.app.server.address().port + '/getPassword', {
+                    dataType: 'json',
+                    contentType: 'application/x-www-form-urlencoded',
+                    method: 'POST',
+                    timeout: 3000,
+                    data: {
+                        mountpoint: mountpoint,
+                        base: headers.authorization.split(' ')[1],
+                        status: findMountpoint.data.status
+                    }
+                });
+                if (getPassword.data.code == 404) {
+                    this.sendResponse(socket, 401, 'Unauthorized', {});
+                    return;
+                }
+            }
+        }
+        if (find !== 'Request-Type: SIZE') {
+            if (!this.mountData.has(mountpoint)) {
+                this.sendResponse(socket, 404, 'Mountpoint Not Found', {
+                    'Access-Control-Allow-Origin': headers.origin
+                });
+                return;
+            }
+        }
+        if (found == 'Client-Type: Virtual-NTRIP-Client/1.0') {
+            const uid = lines[7].split(': ')[1];
+            const token = lines[6].split(': ')[1];
+            let decryptUid = CryptoJS.AES.decrypt(uid, KEY, {
+                mode: CryptoJS.mode.ECB,
+                padding: CryptoJS.pad.Pkcs7
+            }).toString(CryptoJS.enc.Utf8).split('?')[0];
+            const redisType = {
+                "B": "db0",
+                "C": "db1",
+                "T": "db5",
+            };
+            const lastCode = decryptUid[decryptUid.length - 1];
+            const rdmcode = await this.redis.get(redisType[lastCode]).get(token);
+            if (!rdmcode) {
+                this.sendResponse(socket, 401, 'Unauthorized', {
+                    'Access-Control-Allow-Origin': headers.origin
+                });
+                return;
+            }
+        }
+        if (find == 'Request-Type: SIZE') {
+            const uid = lines[8].split(': ')[1];
+            // 接受客户端连接
+            this.sizeSockets.set(uid + ':' + mountpoint, socket);
+            const size = [...this.clientSockets.values()].filter(value => value == mountpoint);
+            this.clientSendResponse(socket, 200, 'OK', {
+                'Access-Control-Allow-Origin': headers.origin, // 设置允许的源
+                'Content-Type': 'text/event-stream',
+                'Connection': 'keep-alive'
+            }, size.length);
+            console.log(`[Caster] 已发送Client挂载数量`);
+        } else {
+            // 接受客户端连接
+            this.clientSockets.set(socket, mountpoint);
+            this.clientSendResponse(socket, 200, 'OK', {
+                'Access-Control-Allow-Origin': headers.origin, // 设置允许的源
+                'Content-Type': 'text/event-stream',
+                'Connection': 'keep-alive'
+            });
+            this.broadSizeData();
+            console.log(`[Caster] Client 已连接至: ${mountpoint}`);
+        }
+    }
+
+    // 广播已连接到此挂载点的客户端数量数据给所有已连接到此挂载点的客户端
+    broadSizeData() {
+        for (const [clientMsg, client] of this.sizeSockets.entries()) {
+            const mountpoint = clientMsg.split(':')[1];
+            const size = [...this.clientSockets.values()].filter(value => value == mountpoint);
+            client.write(size.length + '');
+        }
+    }
+
+    // 广播数据给所有连接到此挂载点的客户端
+    broadcastData(mountpoint, data) {
+        for (const [client, clientMount] of this.clientSockets.entries()) {
+            if (clientMount === mountpoint && !client.destroyed) {
+                client.write(data);
+            }
+        }
+    }
+
+    removeClient(socket) {
+        this.clientSockets.delete(socket);
+    }
+
+    findHeader(lines, headerName) {
+        for (const line of lines) {
+            if (line.startsWith(headerName + ':')) {
+                return line.substring(headerName.length + 1).trim();
+            }
+        }
+        return null;
+    }
+}
+
+// 导出类供单独使用
+module.exports = { SimpleNtripCaster };
--- a/app/middleware/utils/log.js
+++ b/app/middleware/utils/log.js
@@ -0,0 +1,72 @@
+// 错误日志
+const fs = require('fs');
+const os = require('os');
+const path = require('path');
+
+const dirCheck = (dirPath) => {
+    // 检查文件夹是否存在
+    fs.access(dirPath, fs.constants.F_OK, (err) => {
+        if (err) {
+            // 文件夹不存在
+            console.log('文件夹不存在，正在创建...');
+            fs.mkdir(dirPath, { recursive: true }, (err) => {
+                if (err) throw err;
+                console.log('文件夹已创建');
+            });
+        } else {
+            // 文件夹存在
+            console.log('文件夹已存在');
+        }
+    });
+}
+
+module.exports = (filePath, msg) => {
+    // 获取时间
+    let now = new Date(),
+    times = {
+        year: now.getFullYear(),
+        month: now.getMonth() + 1,
+        day: now.getDate(),
+        hour: now.getHours(),
+        minute: now.getMinutes(),
+        second: now.getSeconds()
+    };
+    // 获取操作系统类型
+    const type = os.type();
+    console.log(`操作系统类型: ${type}`);
+    filePath = path.join(__dirname, filePath);
+    dirCheck(filePath);
+    const yearPath = filePath + '/' + times.year;
+    dirCheck(yearPath);
+    const monthPath = yearPath + '/' + times.month;
+    dirCheck(monthPath);
+    const dayPath = monthPath + '/' + times.day + '.txt';
+    const logTime = '[' + times.year + '-' + times.month + '-' + times.day + '>>>' + times.hour + ':' + times.minute + ':' + times.second + ']' + '>>>';
+    let crlf;
+    // 判断操作系统类型，设置换行符
+    if (type === 'Windows_NT') {
+        crlf = '\n';
+    } else {
+        crlf = '\r\n';
+    }
+    // 读取文件内容
+    fs.readFile(dayPath, 'utf8', (err, data) => {
+        if (err) {
+            if (err.code === 'ENOENT') {
+                // 文件不存在，直接写入新内容（这里实际上还是使用了覆盖的方式）
+                fs.writeFile(dayPath, logTime + msg + crlf, (err) => {
+                    if (err) throw err;
+                    console.log('文件已创建并写入新内容');
+                });
+            } else {
+                throw err;
+            }
+        } else {
+            // 文件存在，追加新内容
+            fs.appendFile(dayPath, logTime + msg + crlf, (err) => {
+                if (err) throw err;
+                console.log('新内容已追加到文件');
+            });
+        }
+    });
+};