const express = require('express') const session = require('express-session') const bodyParser = require('body-parser') const request = require('request') const nunjucks = require('nunjucks') const sqlite = require('sqlite') const xml2js = require('xml2js') const path = require('path') const toml = require('toml') const fs = require('fs') const uuid = require('uuid/v4') require('express-async-errors') const util = require('util') const get = util.promisify(request.get) const post = util.promisify(request.post) const dev = process.env.NODE_ENV === 'development' // Load Configuration const filename = path.join(__dirname, 'config.toml') let config let cache = { _updated: 0, streamers: {} } try { config = toml.parse(fs.readFileSync(filename)) } catch (e) { console.error(e) process.exit(1) } config = Object.assign({ 'Streaming': { 'Port': '9322', 'Database': 'streaming.db', 'StreamServer': 'https://tv.icynet.eu/live/', 'ServerHost': 'icynet.eu', 'PublishAddress': 'rtmp://{host}:1935/hls-live/{streamer}', 'Secret': 'changeme' }, 'Auth': { 'Server': 'http://localhost:8282', 'Redirect': 'http://localhost:5000/auth/_callback/' }, 'OAuth2': { 'ClientID': '1', 'ClientSecret': 'changeme' } }, config) // Constants const oauthAuth = '{server}/oauth2/authorize?response_type=code&state={state}&redirect_uri={redirect}&client_id={client}&scope=image' const port = parseInt(config['Streaming']['Port']) const streamServer = config['Streaming']['StreamServer'] const streamServerHost = config['Streaming']['ServerHost'] const authServer = config['Auth']['Server'] const oauthRedirect = config['Auth']['Redirect'] const oauthId = config['OAuth2']['ClientID'].toString() const oauthSecret = config['OAuth2']['ClientSecret'] const streamAppName = streamServer.match(/\/([\w-_]+)\/$/)[1] function teval (str, obj) { let res = str + '' for (let key in obj) { if (res.indexOf('{' + key + '}') === -1) continue res = res.replace('{' + key + '}', obj[key]) } return res } // Database const dbPromise = Promise.resolve() .then(() => sqlite.open(path.join(process.cwd(), config['Streaming']['Database']), { Promise, cache: true })) .then(db => db.migrate()) // Setup server let app = express() app.enable('trust proxy', 1) app.use(bodyParser.urlencoded({ extended: false })) app.use(bodyParser.json()) app.disable('x-powered-by') nunjucks.configure('templates', { autoescape: true, express: app }) app.use(session({ secret: config['Streaming']['Secret'], resave: false, saveUninitialized: true, cookie: { secure: !dev, maxAge: 2678400000 // 1 month } })) // Parse stream metrics from the stat.xml file async function pullMetrics (uuid) { let statPath = streamServer + 'stat' if (!cache.stats || cache._updated < Date.now() - 5000) { let { body } = await get(statPath) let rip = await xml2js.parseStringPromise(body) if (!rip.rtmp.server) throw new Error('Invalid response from server.') // Autofind the correct server let rtmpserver = rip.rtmp.server[0].application let rtmpapp for (let i in rtmpserver) { if (rtmpserver[i].name[0] !== streamAppName) continue rtmpapp = rtmpserver[i] } if (!rtmpapp) throw new Error('Invalid response from server.') cache.stats = rtmpapp.live cache._updated = Date.now() } // Extract applicable stream data let forUser for (let i in cache.stats) { if (!cache.stats[i].stream) continue if (cache.stats[i].stream[0].name[0] !== uuid) continue forUser = cache.stats[i].stream[0] } if (!forUser) return null // Generic data object let data = { time: forUser.time[0], bytes: forUser.bytes_in[0], video: null, audio: null } // Add video metadata, if applicable if (forUser.meta[0].video[0] !== '') { data['video'] = { width: forUser.meta[0].video[0].width[0], height: forUser.meta[0].video[0].height[0], frame_rate: forUser.meta[0].video[0].frame_rate[0], codec: forUser.meta[0].video[0].codec[0] } } // Add audio metadata, if applicable if (forUser.meta[0].audio[0] !== '') { data['audio'] = { sample_rate: forUser.meta[0].audio[0].sample_rate[0], channels: forUser.meta[0].audio[0].channels[0], codec: forUser.meta[0].audio[0].codec[0] } } return data } // Handle requests from nginx-rtmp-module app.post('/publish', async (req, res) => { if (!req.body.name) throw new Error('Invalid request.') let db = await dbPromise // Validate stream key let streamer = await db.get('SELECT * FROM channels WHERE key=?', req.body.name) if (!streamer) throw new Error('Invalid stream key.') console.log('Streamer %s has started streaming!', streamer.name) // Generate real publish address for the server let publishAddress = config['Streaming']['PublishAddress'] .replace('{streamer}', streamer.name) .replace('{host}', '127.0.0.1') // Set channel streaming status db.run('UPDATE channels SET live_at=? WHERE id=?', Date.now(), streamer.id) // Redirect the streaming server to the target res.set('Location', publishAddress) res.status(302) res.end() }) app.post('/publish_done', async (req, res) => { if (!req.body.name) throw new Error('Invalid request.') let db = await dbPromise db.run('UPDATE channels SET live_at=NULL, last_stream=? WHERE key=?', Date.now(), req.body.name) res.send('OK') }) // Front-end server // OAuth2 authenticator app.get('/login', async (req, res) => { if (req.session.user) return res.redirect('/') req.session.state = uuid() res.redirect(teval(oauthAuth, { state: req.session.state, redirect: oauthRedirect, client: oauthId, server: authServer })) }) app.get('/auth/_callback', async (req, res) => { let state = req.session.state if (!state) throw new Error('Something went wrong!') let code = req.query.code let provState = req.query.state if (!code || state !== provState) throw new Error('Something went wrong!') delete req.session.state // Aquire token let { body } = await post(authServer + '/oauth2/token', { form: { grant_type: 'authorization_code', code: code, redirect_uri: oauthRedirect, client_id: oauthId, client_secret: oauthSecret }, auth: { user: oauthId, pass: oauthSecret } }) if (!body) throw new Error('Could not obtain access token!') try { body = JSON.parse(body) } catch (e) { console.error(e, body) throw new Error('Authorization server gave us an invalid response!') } if (body['error']) { throw new Error(body['error'] + ': ' + body['error_description']) } let token = body.access_token // Get user information let { body: bodyNew } = await get(authServer + '/oauth2/user', { auth: { bearer: token } }) try { bodyNew = JSON.parse(bodyNew) } catch (e) { console.error(e, bodyNew) throw new Error('Authorization server gave us an invalid response for user!') } // Get user from database let db = await dbPromise let user = await db.get('SELECT * FROM signed_users WHERE uuid=?', bodyNew.uuid) if (!user) { await db.run('INSERT INTO signed_users (uuid,name) VALUES (?,?)', bodyNew.uuid, bodyNew.username) } req.session.login = bodyNew.uuid req.session.username = bodyNew.username // Lets see if this user is a streamer let streamer = await db.get('SELECT * FROM channels WHERE user_uuid = ?', bodyNew.uuid) if (streamer) cache.streamers[bodyNew.uuid] = streamer res.redirect('/') }) app.get('/logout', (req, res) => { req.session.destroy() res.redirect('/') }) // Views app.use('/dist', express.static(path.join(__dirname, 'dist'), { maxAge: dev ? 0 : 2678400000 })) app.use(async function (req, res, next) { req.isStreamer = false if (!req.session.login) return next() res.locals.session = { uuid: req.session.login, username: req.session.username } if (cache.streamers[req.session.login]) { req.isStreamer = true return next() } next() }) // Index app.get('/', (req, res) => { res.render('index.html', { streamer: req.isStreamer }) }) // Dashboard app.get('/dashboard', (req, res, next) => { if (!req.isStreamer) return next(new Error('Unauthorized')) let stream = cache.streamers[req.session.login] res.render('dashboard.html', { stream: stream.key, server: 'rtmp://' + streamServerHost + '/live/' }) }) // Stats app.get('/dashboard/stats', async (req, res) => { if (!req.isStreamer) return res.jsonp({ error: 'Unauthorized' }) let stream = cache.streamers[req.session.login] let data try { data = await pullMetrics(stream.key) } catch (e) { return res.jsonp({ error: e.message }) } if (!data) return res.jsonp({ error: 'No data was returned.' }) res.jsonp(data) }) // Data app.get('/dashboard/data', async (req, res) => { if (!req.isStreamer) return res.jsonp({ error: 'Unauthorized' }) let stream = cache.streamers[req.session.login] let data let db = await dbPromise try { data = await db.get('SELECT * FROM channels WHERE key=?', stream.key) } catch (e) { return res.jsonp({ error: 'Unauthorized' }) } if (!data) return res.jsonp({ error: 'Unauthorized' }) res.jsonp({ 'name': data.name, 'key': stream.key, 'uuid': req.session.login, 'live': data.live_at != null, 'live_at': new Date(parseInt(data.live_at)), 'last_stream': new Date(parseInt(data.last_stream)) }) }) // Player app.get('/watch/:name', (req, res) => { res.render('player.html', { name: req.params.name, server: streamServer }) }) app.get('/player/:name', (req, res) => { res.redirect('/watch/' + req.params.name) }) // Public data app.get('/api/channel/:name', async (req, res) => { let name = req.params.name let db = await dbPromise let data = await db.get('SELECT name,live_at,last_stream FROM channels WHERE name=?', name) if (!data) return res.jsonp({ error: 'No such channel!' }) data.live = data.live_at != null data.live_at = new Date(parseInt(data.live_at)) data.last_stream = new Date(parseInt(data.last_stream)) res.jsonp(data) }) // Error handler app.use((error, req, res, next) => { if (dev) console.error(error.stack) res.send(error.message) }) // Start server const host = dev ? '0.0.0.0' : '127.0.0.1' app.listen(port, host, () => console.log('Listening on %s:%d', host, port))