2019-10-23 08:24:20 +00:00
|
|
|
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')
|
2019-10-23 08:38:14 +00:00
|
|
|
const connectSession = require('connect-redis')
|
2019-10-23 08:24:20 +00:00
|
|
|
|
|
|
|
require('express-async-errors')
|
|
|
|
|
2019-10-23 08:38:14 +00:00
|
|
|
const SessionStore = connectSession(session)
|
|
|
|
|
2019-10-23 08:24:20 +00:00
|
|
|
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({
|
2019-10-23 08:38:14 +00:00
|
|
|
key: 'Streamserver Session',
|
2019-10-23 08:24:20 +00:00
|
|
|
secret: config['Streaming']['Secret'],
|
|
|
|
resave: false,
|
|
|
|
saveUninitialized: true,
|
2019-10-23 08:38:14 +00:00
|
|
|
store: new SessionStore(),
|
2019-10-23 08:24:20 +00:00
|
|
|
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))
|