1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
|
// live/: a coinos name's owner sends a recording over the WebSocket and it
// comes out as HLS — a live window and the whole of it — with the viewers
// counted; a sender that drops comes back to the same stream (a
// discontinuity between); the end closes the playlist; nobody else may send.
// Run: bun tools/live-ingest-test.js (needs ffmpeg)
import { generateSecretKey, getPublicKey, finalizeEvent } from 'nostr-tools/pure';
import { mkdtempSync, rmSync, readFileSync, writeFileSync, existsSync, readdirSync } from 'node:fs';
import { tmpdir } from 'node:os';
const sleep = (ms) => new Promise((r) => setTimeout(r, ms));
let ok = true;
const check = (n, c, d = '') => { console.log(` ${c ? '✓' : '✗'} ${n}${d ? ' — ' + d : ''}`); if (!c) ok = false; };
const dir = mkdtempSync(tmpdir() + '/live-ingest-');
// what a browser's MediaRecorder sends: WebM, VP8 + Opus, a portrait camera
Bun.spawnSync(['ffmpeg', '-loglevel', 'error', '-y', '-f', 'lavfi', '-i', 'testsrc=size=720x1280:rate=30', '-f', 'lavfi', '-i', 'sine=frequency=440',
'-t', '12', '-c:v', 'libvpx', '-deadline', 'realtime', '-b:v', '1M', '-c:a', 'libopus', '-f', 'webm', dir + '/cam.webm']);
const webm = readFileSync(dir + '/cam.webm');
const SK = generateSecretKey(), PK = getPublicKey(SK), OTHER = generateSecretKey(), ADMIN = generateSecretKey();
const NEWBIE = generateSecretKey();
const names = Bun.serve({ port: 0, hostname: '127.0.0.1', fetch: (req) => {
const n = new URL(req.url).pathname.split('/').pop();
return Response.json(n === 'alice' ? { taken: true, pubkey: PK, manager: null, created: Date.now() - 3 * 86400_000 }
: n === 'newbie' ? { taken: true, pubkey: getPublicKey(NEWBIE), created: Date.now() - 3600_000 } : { taken: false });
} });
writeFileSync(dir + '/config.json', JSON.stringify({ admins: [getPublicKey(ADMIN)] }));
process.env.LIVE_CONFIG = dir + '/config.json';
process.env.LIVE_DATA = dir + '/data';
process.env.NAMES_URL = names.url.origin;
process.env.LIVE_BASE = 'http://x/live';
const live = await import('../live/server.js');
const srv = Bun.serve({ port: 0, hostname: '127.0.0.1', fetch: live.handle, websocket: live.websocket });
const HTTP = `http://127.0.0.1:${srv.port}/live`, WS = `ws://127.0.0.1:${srv.port}/live/ingest`;
const auth = (sk, method = 'GET') => Buffer.from(JSON.stringify(finalizeEvent({ kind: 27235, created_at: Math.floor(Date.now() / 1000), content: '',
tags: [['u', 'https://v3.coinos.io/live/ingest'], ['method', method]] }, sk))).toString('base64');
// open a sender; resolves with its first message (or the refusal)
const connect = (q) => new Promise((resolve) => {
const ws = new WebSocket(`${WS}?${q}`);
ws.binaryType = 'arraybuffer';
ws.onmessage = (e) => resolve({ ws, hello: JSON.parse(e.data) });
ws.onerror = () => resolve({ ws, error: true });
ws.onclose = (e) => resolve({ ws, closed: e.code });
});
// the recording, in half-second chunks at the pace it was made
async function send(ws, from = 0, to = webm.length) {
const step = Math.ceil(webm.length / 24);
for (let i = from; i < to; i += step) { ws.send(webm.subarray(i, Math.min(to, i + step))); await sleep(500); }
}
const text = async (u) => { const r = await fetch(u); return { status: r.status, body: await r.text() }; };
const segs = (m3u8) => m3u8.split('\n').filter((l) => /\.ts$/.test(l));
try {
console.log('[who may send]');
check('a stranger to the name is refused', (await connect(`auth=${auth(OTHER)}&name=alice`)).error === true);
check('...as is a name nobody owns', (await connect(`auth=${auth(SK)}&name=bob`)).error === true);
check('...and a forged auth', (await connect(`auth=${auth(SK).slice(0, -8)}AAAAAAAA&name=alice`)).error === true);
const nip98 = (sk, path, method = 'GET') => 'Nostr ' + Buffer.from(JSON.stringify(finalizeEvent({ kind: 27235, created_at: Math.floor(Date.now() / 1000), content: '',
tags: [['u', 'https://v3.coinos.io' + path], ['method', method]] }, sk))).toString('base64');
check('...and a name claimed an hour ago (a day first)', (await connect(`auth=${auth(NEWBIE)}&name=newbie`)).error === true);
const al = await (await fetch(`${HTTP}/allowed?name=newbie`, { headers: { authorization: nip98(NEWBIE, '/live/allowed') } })).json();
check('...and asked, the server says when it can', al.ok === false && al.reason === 'new-name' && al.until > Date.now() + 22 * 3600_000, JSON.stringify(al));
check('a day-old name\'s owner may', (await (await fetch(`${HTTP}/allowed?name=alice`, { headers: { authorization: nip98(SK, '/live/allowed') } })).json()).ok === true);
console.log('\n[TURN for calls]');
const turnAuth = (sk, path = '/live/turn') => 'Nostr ' + Buffer.from(JSON.stringify(finalizeEvent({ kind: 27235, created_at: Math.floor(Date.now() / 1000), content: '',
tags: [['u', 'https://v3.coinos.io' + path], ['method', 'GET']] }, sk))).toString('base64');
check('ICE servers need a signed ask', (await fetch(`${HTTP}/turn`)).status === 401);
check('...one signed for this address', (await fetch(`${HTTP}/turn`, { headers: { authorization: turnAuth(OTHER, '/live/ingest') } })).status === 401);
const ice = await (await fetch(`${HTTP}/turn`, { headers: { authorization: turnAuth(OTHER) } })).json();
check('...anyone signed gets them (STUN alone with no TURN key configured)', Array.isArray(ice.iceServers) && /^stun:/.test(ice.iceServers[0].urls[0]), JSON.stringify(ice));
console.log('\n[sending]');
const a = await connect(`auth=${auth(SK)}&name=alice`);
check('the owner is let in and told where the stream is', !!a.hello?.id && /\/live\/[0-9a-f]{16}\/live\.m3u8$/.test(a.hello.live) && /index\.m3u8$/.test(a.hello.recording), JSON.stringify(a.hello));
const id = a.hello.id;
check('...and can\'t start a second one at once', (await connect(`auth=${auth(SK)}&name=alice`)).error === true);
await send(a.ws, 0, Math.floor(webm.length * 0.6));
await sleep(2500);
const l1 = await text(`${HTTP}/${id}/live.m3u8`);
check('the live playlist has segments, and no end', l1.status === 200 && segs(l1.body).length >= 2 && !/ENDLIST/.test(l1.body), segs(l1.body).length + ' segments');
const seg = await fetch(`${HTTP}/${id}/${segs(l1.body)[0]}`);
const b = new Uint8Array(await seg.arrayBuffer());
check('a segment is MPEG-TS', seg.status === 200 && b[0] === 0x47, seg.status + ' ' + b.length + ' bytes');
const probe = JSON.parse(Bun.spawnSync(['ffprobe', '-v', 'error', '-show_streams', '-of', 'json', `${dir}/data/${id}/${segs(l1.body)[0]}`]).stdout.toString()).streams;
const v = probe.find((x) => x.codec_type === 'video'), au = probe.find((x) => x.codec_type === 'audio');
check('...H.264 at most 1280 on its long side, AAC sound', v?.codec_name === 'h264' && Math.max(v.width, v.height) <= 1280 && v.height > v.width && au?.codec_name === 'aac', v && `${v.codec_name} ${v.width}x${v.height} ${au?.codec_name}`);
const st = await (await fetch(`${HTTP}/${id}/stats`)).json();
check('the watcher who fetched the playlist is counted', st.viewers === 1 && st.state === 'live', JSON.stringify(st));
// WebRTC (Cloudflare's SFU) needs its key in the config: here there's none, and the app keeps to HLS
check('with no SFU key, no WebRTC: stats say so, and watch/publish answer 501', st.rtc === null
&& (await fetch(`${HTTP}/${id}/rtc/watch`, { method: 'POST' })).status === 501 && (await fetch(`${HTTP}/${id}/rtc/publish`, { method: 'POST', body: '{}' })).status === 501);
console.log('\n[a sender that drops comes back]');
a.ws.close();
await sleep(500);
check('the stream waits for it', (await (await fetch(`${HTTP}/${id}/stats`)).json()).state === 'waiting');
const b2 = await connect(`auth=${auth(SK)}&name=alice&resume=${id}`);
check('...it carries on the same stream', b2.hello?.id === id && b2.hello.resumed === true, JSON.stringify(b2.hello));
await send(b2.ws, 0, webm.length); // a new recording starts from its own header
await sleep(2500);
const full = await text(`${HTTP}/${id}/index.m3u8`);
check('the whole playlist joins the two with a discontinuity', /#EXT-X-DISCONTINUITY\n/.test(full.body) && segs(full.body).some((x) => x.startsWith('0-')) && segs(full.body).some((x) => x.startsWith('1-')), segs(full.body).length + ' segments');
check('...as an EVENT playlist while live', /PLAYLIST-TYPE:EVENT/.test(full.body) && !/ENDLIST/.test(full.body));
console.log('\n[reports and taking down]');
const post = (sk, path, body) => fetch(`http://127.0.0.1:${srv.port}${path}`, { method: 'POST', headers: { authorization: nip98(sk, path, 'POST'), 'content-type': 'application/json' }, body: JSON.stringify(body) });
check('a viewer reports the stream', (await post(OTHER, `/live/${id}/report`, { reason: 'sexual', text: 'look' })).ok);
const reps = readFileSync(dir + '/data/.reports.jsonl', 'utf8').trim().split('\n').map((l) => JSON.parse(l));
check('...kept, with who and why', reps.length === 1 && reps[0].stream === id && reps[0].reason === 'sexual' && reps[0].by === getPublicKey(OTHER));
check('...unsigned, refused', (await fetch(`${HTTP}/${id}/report`, { method: 'POST', body: '{}' })).status === 401);
check('only an admin can take it down', (await post(OTHER, `/live/${id}/remove`, { ban: true })).status === 403);
console.log('\n[the end]');
b2.ws.send('end');
await sleep(1500);
const done = await text(`${HTTP}/${id}/index.m3u8`);
check('ending closes the playlist: the recording', /ENDLIST/.test(done.body) && /PLAYLIST-TYPE:VOD/.test(done.body));
check('...the live one ends too', /ENDLIST/.test((await text(`${HTTP}/${id}/live.m3u8`)).body));
const total = Math.round([...done.body.matchAll(/#EXTINF:([\d.]+)/g)].reduce((n, m) => n + +m[1], 0));
check('...about the length sent (12 s + 7 s)', total >= 14 && total <= 24, total + ' s');
check('a stream that never was: 404', (await fetch(`${HTTP}/0123456789abcdef/live.m3u8`)).status === 404);
const c = await connect(`auth=${auth(SK)}&name=alice`);
check('once ended, the owner can go live again', !!c.hello?.id && c.hello.id !== id);
c.ws.send('end');
await sleep(500);
console.log('\n[the streamer deletes a recording]');
const cid = c.hello.id;
check('someone else can\'t delete it', (await post(OTHER, `/live/${cid}/delete`, {})).status === 403);
const del = await (await post(SK, `/live/${cid}/delete`, {})).json();
check('its streamer can', del.deleted === true, JSON.stringify(del));
check('...its playlist is gone (410), said to be deleted', (await fetch(`${HTTP}/${cid}/index.m3u8`)).status === 410 && (await (await fetch(`${HTTP}/${cid}/stats`)).json()).state === 'deleted');
check('...its files with it (only a note of it left)', readdirSync(`${dir}/data/${cid}`).join() === 'stream.json');
check('...and the owner can still go live', (await (await fetch(`${HTTP}/allowed?name=alice`, { headers: { authorization: nip98(SK, '/live/allowed') } })).json()).ok === true);
const kept = await (await post(SK, `/live/${id}/delete`, {})).json();
check('a reported one, deleted by its streamer, is gone from view but kept aside for reporting', kept.deleted === true
&& (await fetch(`${HTTP}/${id}/index.m3u8`)).status === 410 && existsSync(`${dir}/data/.removed/${id}`) && !existsSync(`${dir}/data/${id}`));
const c0 = await connect(`auth=${auth(SK)}&name=alice`);
const tid = c0.hello.id;
await send(c0.ws, 0, Math.floor(webm.length * 0.3));
await sleep(1500);
const rm = await (await post(ADMIN, `/live/${tid}/remove`, { ban: true })).json();
check('the admin takes a stream down, banning its sender', rm.removed === true && rm.banned === true, JSON.stringify(rm));
check('...its playlists are gone (410), its segments too', (await fetch(`${HTTP}/${tid}/live.m3u8`)).status === 410 && (await fetch(`${HTTP}/${tid}/0-00000.ts`)).status === 410);
check('...kept aside for reporting, not deleted', existsSync(`${dir}/data/.removed/${tid}`) && !existsSync(`${dir}/data/${tid}`));
check('...and its sender can\'t go live again', (await connect(`auth=${auth(SK)}&name=alice`)).error === true
&& (await (await fetch(`${HTTP}/allowed?name=alice`, { headers: { authorization: nip98(SK, '/live/allowed') } })).json()).reason === 'banned');
} catch (e) { console.error(e); ok = false; } finally {
srv.stop(true); names.stop(true);
await sleep(300);
rmSync(dir, { recursive: true, force: true });
}
console.log(ok ? '\n✅ live: sent in, watched as HLS' : '\n❌ failed');
process.exit(ok ? 0 : 1);
|
