flexiwan项目踩坑实践

项目背景

flexiManage是以色列一家初创公司flexiWAN开源的基于SD-WAN平台的应用层的框架，包括flexiManage服务端框架，基于此服务端框架进行了一些借鉴和改进

目录结构

• api
• billing
• bin
• broker
• controllers
• deviceLogic
• logging
• logs
• migrations
• models
• notifications
• periodic
• public
• routes
• services
• utils
• websocket
• authenticate.js
• configs.js
• expressserver.js
• flexibilling.js
• mongoConns.js
• rateLimitStore.js
• token.js

踩坑案例

BFF抹掉https的node模块验证

[bug描述] 做验证使用服务端及硬件侧未配置ssl，而node启动https模块会默认验证ssl，导致无法启动服务

[bug分析] node模块的ssl验证

[解决方案] 起一层bff用于透传接口，后续方便将后续服务层进行微服务化等处理

process.env.NODE_TLS_REJECT_UNAUTHORIZED = '0';
const express = require('express');
const request = require('request');
const app = express();
const bodyParser = require('body-parser');
const router = express.Router();

const SUCC_REG = /^2[0-9]{2}$/

app.use(bodyParser.urlencoded({ extended: false }));
app.use(bodyParser.json());

const headers = {
            'authorization': "Bearer eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.eyJfaWQiOiI1ZmEzYTY5OGZjNDI2ODEwODc3MDYzZDQiLCJ1c2VybmFtZSI6Im1jYWlkYW9Ac2luYS5jb20iLCJvcmciOiI1ZmFkZTkyZDljNGQ2MDQyOWRjN2RhNmMiLCJvcmdOYW1lIjoidHQiLCJhY2NvdW50IjoiNWZhM2E2OThmYzQyNjgxMDg3NzA2M2QzIiwiYWNjb3VudE5hbWUiOiJ0ZXN0IiwicGVybXMiOnsiam9icyI6MTUsImJpbGxpbmciOjMsImFjY291bnRzIjo3LCJvcmdhbml6YXRpb25zIjoxNSwiZGV2aWNlcyI6MTUsInRva2VucyI6MTUsImFwcGlkZW50aWZpY2F0aW9ucyI6MTUsIm1lbWJlcnMiOjE1LCJ0dW5uZWxzIjoxNSwiYWNjZXNzdG9rZW5zIjoxNSwibm90aWZpY2F0aW9ucyI6MTUsInBhdGhsYWJlbHMiOjE1LCJtbHBvbGljaWVzIjoxNX0sImlhdCI6MTYwODExMjcwMiwiZXhwIjoxNjA4NzE3NTAyfQ.LYFv1pBP1540gb-NRCCe4dvbQ0T9HSoZHMkD8xkMFLc",
            'Content-Type': 'application/json'
        },
        errMsg = {
            msg:'unexpected response'
        },
        baseUrl = 'https://10.100.37.101:3443';


// 获取所有设备接口
app.get('/api/devices',(req,res)=> {
    console.log(req.url)
    request({
        url: `${baseUrl}${req.url}`,
        method: 'GET',
        headers
    }, (err, response, body) => {
        console.log(response.statusCode)
        if(SUCC_REG.test(response.statusCode)) {
            res.send({code: 200,msg:JSON.parse(response.body)})
        } else {
            res.send(errMsg)
        }
    })
});

// 获取单个设备接口
app.get('/api/devices/:id',(req,res)=> {
    console.log(req.url)
    request({
        url: `${baseUrl}${req.url}`,
        method: 'GET',
        headers
    }, (err, response, body) => {
        console.log(response.statusCode)
        if(SUCC_REG.test(response.statusCode)) {
            res.send({code: 200,msg:JSON.parse(response.body)})
        } else {
            res.send(errMsg)
        }
    })
});

// 获取路由接口
app.get('/api/devices/:id/routes',(req,res)=> {
    console.log(req.url)
    request({
        url: `https://10.100.37.101:3443/api/devices/${req.params.id}/routes`,
        method: 'GET',
        headers
    }, (err, response, body) => {
        console.log(response.statusCode)
        if(SUCC_REG.test(response.statusCode)) {
            res.send({code: 200,msg:JSON.parse(response.body)})
        } else {
            res.send(errMsg)
        }
    })
});

// 启动单个设备
app.post('/api/devices/:id/apply/start',(req,res)=> {
    console.log(req.url);
    request({
        url: `${baseUrl}/api/devices/${req.params.id}/apply`,
        method: 'POST',
        headers,
        body: JSON.stringify({
            "method": "start"
        })
    }, (err, response, body) => {
        let r = JSON.parse(body)
        if(r.status == 'completed') {
            res.send({code: 200,msg:'start success'})
        } else {
            res.send({msg: 'start error'})
        }
    })
});

// 停止单个设备
app.post('/api/devices/:id/apply/stop',(req,res)=> {
    console.log(req.url)
    request({
        url: `${baseUrl}/api/devices/${req.params.id}/apply`,
        method: 'POST',
        headers,
        body: JSON.stringify({
            "method": "stop"
        })
    }, (err, response, body) => {
        let r = JSON.parse(body)
        if(r.status == 'completed') {
            res.send({code: 200,msg:'stop success'})
        } else {
            res.send({msg: 'stop error'})
        }
    })
});

// 同步单个设备
app.post('/api/devices/:id/apply',(req,res)=> {
    console.log(req.url)
    request.post({
        url: `${baseUrl}${req.url}`,
        headers,
        body: JSON.stringify({
            "method": "sync"
        })
    }, (err, response, body) => {
        let r = JSON.parse(body)
        if(r.status == 'completed') {
            res.send({code: 200,msg:'update success'})
        } else {
            res.send({msg: 'update error'})
        }
    })
});

// 删除单个设备
app.delete('/api/devices/:id',(req,res)=> {
    console.log(req.url)
    request({
        url: `${baseUrl}${req.url}`,
        method: 'DELETE',
        headers
    }, (err, response, body) => {
        console.log(response.statusCode)
        if(SUCC_REG.test(response.statusCode)) {
            res.send({code: 200,msg:JSON.parse(response.body)})
        } else {
            res.send(errMsg)
        }
    })
});

// 更新设备详情
app.put('/api/devices/:id',(req,res)=> {
    request({
        url: `${baseUrl}${req.url}`,
        method: 'PUT',
        headers,
        body: JSON.stringify(req.body)
    }, (err, response, body) => {
        console.log('put device', response.statusCode)
        if(SUCC_REG.test(response.statusCode)) {
            res.send({code: 200,msg:JSON.parse(response.body)})
        } else {
            res.send(errMsg)
            console.log('error device', response.statusCode, response.body)
        }
    })
});

// 删除隧道接口
app.post('/api/devices/apply/delTunnel',(req,res)=> {
    console.log('req.body', req.body)
    request.post({
        url: `${baseUrl}/api/devices/apply`,
        headers,
        body: JSON.stringify(req.body)
    }, (err, response, body) => {
        let r = JSON.parse(body)
        console.log(r)
        if(r.status == 'completed') {
            res.send({code: 200,msg:'删除隧道成功'})
        } else {
            res.send({msg: r.error})
        }
    })
});

// 建立隧道接口
app.post('/api/devices/apply/createTunnel',(req,res)=> {
    console.log(req.body)
    request.post({
        url: `${baseUrl}/api/devices/apply`,
        headers,
        body: JSON.stringify(req.body)
    }, (err, response, body) => {
        let r = JSON.parse(body)
        console.log(r)
        if(r.status == 'completed') {
            res.send({code: 200,msg:r.message})
        } else {
            res.send({msg: r.error})
        }
    })
});



// 获取所有隧道接口
app.get('/api/tunnels',(req,res)=> {
    console.log(req.url)
    request({
        url: `${baseUrl}${req.url}`,
        method: 'GET',
        headers
    }, (err, response, body) => {
        console.log(response.statusCode)
        if(SUCC_REG.test(response.statusCode)) {
            res.send({code: 200,msg:JSON.parse(response.body)})
        } else {
            res.send(errMsg)
        }
    })
});

app.listen(6000, '127.0.0.1', ()=>{
    console.log('app server');
});

express请求接口请求体不同无法匹配

[bug描述] express实例中同样post请求，只是body体不同而导致无法区分，从而覆盖后续接口

[bug分析] express的中间件原理，在加载路由过程正则匹配后不会匹配body体

[解决方案] 区分路由接口，通过request转发或加上路由模块区分

// 启动单个设备
app.post('/api/devices/:id/apply/start',(req,res)=> {
    console.log(req.url);
    request({
        url: `${baseUrl}/api/devices/${req.params.id}/apply`,
        method: 'POST',
        headers,
        body: JSON.stringify({
            "method": "start"
        })
    }, (err, response, body) => {
        let r = JSON.parse(body)
        if(r.status == 'completed') {
            res.send({code: 200,msg:'start success'})
        } else {
            res.send({msg: 'start error'})
        }
    })
});

// 停止单个设备
app.post('/api/devices/:id/apply/stop',(req,res)=> {
    console.log(req.url)
    request({
        url: `${baseUrl}/api/devices/${req.params.id}/apply`,
        method: 'POST',
        headers,
        body: JSON.stringify({
            "method": "stop"
        })
    }, (err, response, body) => {
        let r = JSON.parse(body)
        if(r.status == 'completed') {
            res.send({code: 200,msg:'stop success'})
        } else {
            res.send({msg: 'stop error'})
        }
    })
});

源码解析

主要是以express为核心的node应用，封装了express的基类进行实例，配合websocket进行实时数据的连接，redis的输出消费存储

expressserver

class ExpressServer {
  constructor (port, securePort, openApiYaml) {
    this.port = port;
    this.securePort = securePort;
    this.app = express();
    this.openApiPath = openApiYaml;
    this.schema = yamljs.load(openApiYaml);
    const restServerUrl = configs.get('restServerUrl');
    const servers = this.schema.servers.filter(server => server.url.includes(restServerUrl));
    if (servers.length === 0) {
      this.schema.servers.unshift({
        description: 'Local Server',
        url: restServerUrl + '/api'
      });
    }

    this.setupMiddleware = this.setupMiddleware.bind(this);
    this.addErrorHandler = this.addErrorHandler.bind(this);
    this.onError = this.onError.bind(this);
    this.onListening = this.onListening.bind(this);
    this.launch = this.launch.bind(this);
    this.close = this.close.bind(this);

    this.setupMiddleware();
  }

  setupMiddleware () {
    // this.setupAllowedMedia();
    this.app.use((req, res, next) => {
      console.log(`${req.method}: ${req.url}`);
      return next();
    });

    // Request logging middleware - must be defined before routers.
    this.app.use(reqLogger);
    this.app.set('trust proxy', true); // Needed to get the public IP if behind a proxy

    // Don't expose system internals in response headers
    this.app.disable('x-powered-by');

    // Use morgan request logger in development mode
    if (configs.get('environment') === 'development') this.app.use(morgan('dev'));

    // Start periodic device tasks
    deviceStatus.start();
    deviceQueues.start();
    deviceSwVersion.start();
    deviceSwUpgrade.start();
    notifyUsers.start();
    appRules.start();

    // Secure traffic only
    this.app.all('*', (req, res, next) => {
      // Allow Let's encrypt certbot to access its certificate dirctory
      if (!configs.get('shouldRedirectHttps') ||
          req.secure || req.url.startsWith('/.well-known/acme-challenge')) {
        return next();
      } else {
        return res.redirect(
          307, 'https://' + req.hostname + ':' + configs.get('redirectHttpsPort') + req.url
        );
      }
    });

    // Global rate limiter to protect against DoS attacks
    // Windows size of 5 minutes
    const inMemoryStore = new RateLimitStore(5 * 60 * 1000);
    const rateLimiter = rateLimit({
      store: inMemoryStore,
      max: +configs.get('userIpReqRateLimit'), // Rate limit for requests in 5 min per IP address
      message: 'Request rate limit exceeded',
      onLimitReached: (req, res, options) => {
        logger.error(
          'Request rate limit exceeded. blocking request', {
            params: { ip: req.ip },
            req: req
          });
      }
    });
    this.app.use(rateLimiter);

    // General settings here
    this.app.use(cors.cors);
    this.app.use(bodyParser.json());
    this.app.use(express.json());
    this.app.use(express.urlencoded({ extended: false }));
    this.app.use(cookieParser());

    // Routes allowed without authentication
    this.app.use(express.static(path.join(__dirname, configs.get('clientStaticDir'))));

    // Secure traffic only
    this.app.all('*', (req, res, next) => {
      // Allow Let's encrypt certbot to access its certificate dirctory
      if (!configs.get('shouldRedirectHttps') ||
          req.secure || req.url.startsWith('/.well-known/acme-challenge')) {
        return next();
      } else {
        return res.redirect(
          307, 'https://' + req.hostname + ':' + configs.get('redirectHttpsPort') + req.url
        );
      }
    });

    // no authentication
    this.app.use('/api/connect', require('./routes/connect'));
    this.app.use('/api/users', require('./routes/users'));

    // add API documentation
    this.app.use('/api-docs', swaggerUI.serve, swaggerUI.setup(this.schema));

    // initialize passport and authentication
    this.app.use(passport.initialize());

    // Enable db admin only in development mode
    if (configs.get('environment') === 'development') {
      logger.warn('Warning: Enabling UI database access');
      this.app.use('/admindb', mongoExpress(mongoExpressConfig));
    }

    // Enable routes for non-authorized links
    this.app.use('/ok', express.static(path.join(__dirname, 'public', 'ok.html')));
    this.app.use('/spec', express.static(path.join(__dirname, 'api', 'openapi.yaml')));
    this.app.get('/hello', (req, res) => res.send('Hello World'));

    this.app.get('/api/version', (req, res) => res.json({ version }));

    this.app.use(cors.corsWithOptions);
    this.app.use(auth.verifyUserJWT);
    // this.app.use(auth.verifyPermission);

    try {
      // FIXME: temporary map the OLD routes
      // this.app.use('/api/devices', require('./routes/devices'));
      // this.app.use('/api/devicestats', require('./routes/deviceStats'));
      // this.app.use('/api/jobs', require('./routes/deviceQueue'));
      this.app.use('/api/portals', require('./routes/portals'));
    } catch (error) {
      logger.error('Error: Can\'t connect OLD routes');
    }

    // Intialize routes
    this.app.use('/api/admin', adminRouter);

    const validator = new OpenApiValidator({
      apiSpec: this.openApiPath,
      validateRequests: true,
      validateResponses: configs.get('validateOpenAPIResponse')
    });

    validator
      .install(this.app)
      .then(async () => {
        await this.app.use(openapiRouter());
        await this.launch();
        logger.info('Express server running');
      });
  }

  addErrorHandler () {
    // "catchall" handler, for any request that doesn't match one above, send back index.html file.
    this.app.get('*', (req, res, next) => {
      logger.info('Route not found', { req: req });
      res.sendFile(path.join(__dirname, configs.get('clientStaticDir'), 'index.html'));
    });

    // catch 404 and forward to error handler
    this.app.use(function (req, res, next) {
      next(createError(404));
    });

    // Request error logger - must be defined after all routers
    // Set log severity on the request to log errors only for 5xx status codes.
    this.app.use((err, req, res, next) => {
      req.logSeverity = err.status || 500;
      next(err);
    });
    this.app.use(errLogger);

    /**
     * suppressed eslint rule: The next variable is required here, even though it's not used.
     *
     ** */
    // eslint-disable-next-line no-unused-vars
    this.app.use((error, req, res, next) => {
      const errorResponse = error.error || error.message || error.errors || 'Unknown error';
      res.status(error.status || 500);
      res.type('json');
      res.json({ error: errorResponse });
    });
  }

  /**
   * Event listener for HTTP/HTTPS server "error" event.
   */
  onError (port) {
    return function (error) {
      if (error.syscall !== 'listen') {
        throw error;
      }

      const bind = 'Port ' + port;

      // handle specific listen errors with friendly messages
      /* eslint-disable no-unreachable */
      switch (error.code) {
        case 'EACCES':
          console.error(bind + ' requires elevated privileges');
          process.exit(1);
        case 'EADDRINUSE':
          console.error(bind + ' is already in use');
          process.exit(1);
        default:
          throw error;
      }
    };
  }

  /**
  * Event listener for HTTP server "listening" event.
  */
  onListening (server) {
    return function () {
      const addr = server.address();
      const bind = typeof addr === 'string' ? 'pipe ' + addr : 'port ' + addr.port;
      console.debug('Listening on ' + bind);
    };
  }

  async launch () {
    this.addErrorHandler();

    try {
      this.server = http.createServer(this.app);

      this.options = {
        key: fs.readFileSync(path.join(__dirname, 'bin', configs.get('httpsCertKey'))),
        cert: fs.readFileSync(path.join(__dirname, 'bin', configs.get('httpsCert')))
      };
      this.secureServer = https.createServer(this.options, this.app);

      // setup wss here
      this.wss = new WebSocket.Server({
        server: configs.get('shouldRedirectHttps') ? this.secureServer : this.server,
        verifyClient: connections.verifyDevice
      });

      connections.registerConnectCallback('broker', broker.deviceConnectionOpened);
      connections.registerCloseCallback('broker', broker.deviceConnectionClosed);
      connections.registerCloseCallback('deviceStatus', deviceStatus.deviceConnectionClosed);

      this.wss.on('connection', connections.createConnection);
      console.log('Websocket server running');

      this.server.listen(this.port, () => {
        console.log('HTTP server listening on port', { params: { port: this.port } });
      });
      this.server.on('error', this.onError(this.port));
      this.server.on('listening', this.onListening(this.server));

      this.secureServer.listen(this.securePort, () => {
        console.log('HTTPS server listening on port', { params: { port: this.securePort } });
      });
      this.secureServer.on('error', this.onError(this.securePort));
      this.secureServer.on('listening', this.onListening(this.secureServer));
    } catch (error) {
      console.log('Express server lunch error', { params: { message: error.message } });
    }
  }

  async close () {
    if (this.server !== undefined) {
      await this.server.close();
      console.log(`HTTP Server on port ${this.port} shut down`);
    }
    if (this.secureServer !== undefined) {
      await this.secureServer.close();
      console.log(`HTTPS Server on port ${this.securePort} shut down`);
    }
  }
}

封装了一个express的基类，主要包含中间件的处理、错误处理、监听server

总结

基于express封装的扩展应用，主要利用的是express的中间件原理，可以同类类比nest.js，其核心也是基于express封装的应用，但nest.js基于ng的模块思想做的隔离性更好，更像是服务端的一种node版的spring框架，而本应用确实还是像express的node应用，略显冗余