glass beadian
aa5e262f3a
Self-hosted multi-platform broadcast newsroom for activist collectives: publishing pipeline with group approval and DST-proof scheduling, TweetDeck-style column wall with hostile-content-safe rendering, live control room with Auto Director, bridged chat governance with polls and quorum, distributed sandboxed transcode workers, encrypted content-addressed archive, monthly analytics. MIT licensed. See README.md and docs/.
109 lines
8.9 KiB
TypeScript
109 lines
8.9 KiB
TypeScript
import assert from "node:assert/strict";
|
|
import { spawnSync } from "node:child_process";
|
|
import { generateKeyPairSync, randomUUID } from "node:crypto";
|
|
import { mkdirSync, mkdtempSync, readFileSync, rmSync, writeFileSync } from "node:fs";
|
|
import { tmpdir } from "node:os";
|
|
import { join, resolve } from "node:path";
|
|
import test from "node:test";
|
|
import { signCommand } from "../bus/envelope";
|
|
import { DurableFence } from "../bus/fence";
|
|
import type { ObservedStore } from "../bus/redis";
|
|
import { buildCompositorArgs, buildEncoderArgs, buildIngestArgs, buildPlaylistRemuxArgs, buildSlateArgs } from "../compositor/command-builder";
|
|
import { buildHlsTapArgs } from "../preview/argv";
|
|
import { ffmpegArgs, spawnManaged, type ManagedProcess } from "../relay/ffmpeg";
|
|
import type { DesiredChannel } from "../types";
|
|
import { MediaSupervisor, type RelayJob } from "./index";
|
|
import { supervisorSandboxEnabled } from "./sandbox";
|
|
const pkg = (): {
|
|
sandboxCapability(options: { rootfs: string; cgroupRoot: string }): { available: boolean; reason?: string };
|
|
MediaCommandBuilder: new () => { supervisor(binary: "ffmpeg", args: readonly string[], scratch: string): { argv: readonly string[] } };
|
|
} => require(resolve(__dirname, "../../../../packages/media-sandbox/dist"));
|
|
const channel: DesiredChannel = {
|
|
workspaceId: "sandbox-workspace", channelId: "sandbox-channel", workerId: "sandbox-worker", assignmentEpoch: 1,
|
|
desiredVersion: 1, desiredStatus: "RUNNING", profile: "720p30", mode: "passthrough",
|
|
source: { kind: "hls", url: "https://source.example/live.m3u8" },
|
|
output: { endpointUrl: "rtmp://output.example/live", credentialHandleId: "opaque-handle" },
|
|
};
|
|
const store: ObservedStore = { async get() { return null; }, async put() {}, async heartbeat() {} };
|
|
async function env<T>(values: Record<string, string | undefined>, action: () => Promise<T>): Promise<T> {
|
|
const previous = Object.fromEntries(Object.keys(values).map(key => [key, process.env[key]]));
|
|
for (const [key, value] of Object.entries(values)) { if (value === undefined) delete process.env[key]; else process.env[key] = value; }
|
|
try { return await action(); }
|
|
finally { for (const [key, value] of Object.entries(previous)) { if (value === undefined) delete process.env[key]; else process.env[key] = value; } }
|
|
}
|
|
function command(keys: ReturnType<typeof generateKeyPairSync>) {
|
|
return signCommand({ workspaceId: channel.workspaceId, channelId: channel.channelId, workerId: channel.workerId,
|
|
assignmentEpoch: 1, commandId: randomUUID(), desiredVersion: 1,
|
|
issuedAt: new Date(Date.now() - 1000).toISOString(), expiresAt: new Date(Date.now() + 60_000).toISOString(), snapshot: channel }, "control", keys.privateKey);
|
|
}
|
|
test("sandbox flag defaults locally off and on in production; production off is rejected", async () => {
|
|
await env({ NODE_ENV: "development", MEDIA_SANDBOX: undefined }, async () => assert.equal(supervisorSandboxEnabled(), false));
|
|
await env({ NODE_ENV: "production", MEDIA_SANDBOX: undefined }, async () => assert.equal(supervisorSandboxEnabled(), true));
|
|
await env({ NODE_ENV: "production", MEDIA_SANDBOX: "off" }, async () => assert.throws(supervisorSandboxEnabled, /requires/));
|
|
});
|
|
test("sandbox builder accepts existing relay/ingest/compositor/encoder/playlist/preview plans", () => {
|
|
const builder = new (pkg().MediaCommandBuilder)(); const canvas = { width: 1280, height: 720, fps: 30 };
|
|
const plans = [ffmpegArgs(channel), buildPlaylistRemuxArgs(0),
|
|
buildIngestArgs({ canvas, input: { kind: "mpegtsPipe" }, audio: "source" }),
|
|
buildIngestArgs({ canvas, input: { kind: "synthetic", pattern: "testsrc2", seed: 1, toneHz: 440 }, audio: "source" }),
|
|
buildEncoderArgs({ canvas, videoBitrateKbps: 2800, audioBitrateKbps: 128, output: { kind: "null" }, programTap: { videoFd: 5, audioFd: 6, size: { width: 320, height: 180 } } }),
|
|
buildSlateArgs({ canvas, seconds: 5, image: `/job/${"a".repeat(64)}.png`, offsetUs: 0 }),
|
|
buildHlsTapArgs({ geometry: canvas, rendition: "video", outDir: "/job/taps/video" }),
|
|
buildCompositorArgs({ sources: [{ video: { kind: "rawPipe", fd: 3 }, audio: { kind: "rawPipe", fd: 4 } }], assets: [`/job/${"a".repeat(64)}.png`],
|
|
graph: { canvas, layout: "single", tiles: [{ source: 0, rect: { x: 0, y: 0, w: 1280, h: 720 }, fit: "cover" }],
|
|
overlays: [{ kind: "chyron", asset: 0, y: 10, speedPxPerSec: 40 }], audio: { inputs: [{ source: 0, gainDb: 0 }], masterGainDb: 0 } }, output: { kind: "gate", videoFd: 1, audioFd: 3 } })];
|
|
for (const args of plans) assert.ok(builder.supervisor("ffmpeg", args, "/job").argv.includes("-protocol_whitelist"));
|
|
});
|
|
test("supervisor fails closed if sandbox is enabled without configuration", async () => {
|
|
const dir = mkdtempSync(join(tmpdir(), "media-sandbox-closed-")); const keys = generateKeyPairSync("ed25519");
|
|
const lines: string[] = [];
|
|
await env({ NODE_ENV: "development", MEDIA_SANDBOX: "on", MEDIA_SANDBOX_ROOTFS: undefined, MEDIA_SANDBOX_CGROUP_ROOT: undefined, MEDIA_SCRATCH_DIR: join(dir, "scratch") }, async () => {
|
|
const worker = new MediaSupervisor(channel.workerId, [channel], store, new DurableFence(join(dir, "fence")), new Map([["control", keys.publicKey]]),
|
|
(): RelayJob => ({ async start() { spawnManaged("ffmpeg", ffmpegArgs(channel), ["ignore", "ignore", "pipe"], { redact: v => v }, line => lines.push(line)); }, async stop() {} }),
|
|
{ transition: event => lines.push(JSON.stringify(event)) }, { schedule() { return { cancel() {} }; } });
|
|
try { await worker.start(); await worker.handle(command(keys)); assert.equal(worker.status()[0]?.state, "RESTARTING"); assert.ok(lines.join("\n").includes("SandboxUnavailable")); }
|
|
finally { await worker.stop(); }
|
|
});
|
|
rmSync(dir, { recursive: true, force: true });
|
|
});
|
|
|
|
for (const enabled of [false, true]) test(`secret-leak canaries through signed supervisor command (sandbox ${enabled ? "on" : "off"})`, { timeout: 20_000 }, async t => {
|
|
const dir = mkdtempSync(join(tmpdir(), "media-supervisor-canary-"));
|
|
const rootfs = join(dir, "rootfs");
|
|
for (const d of ["upload", "scratch", "tmp", "proc", "dev", "usr/bin"]) mkdirSync(join(rootfs, d), { recursive: true });
|
|
writeFileSync(join(rootfs, "dev/null"), "");
|
|
try {
|
|
if (enabled) {
|
|
const compiled = spawnSync("cc", ["-static", "-O0", resolve(__dirname, "../../../../packages/media-sandbox/native/test-child.c"), "-o", join(rootfs, "usr/bin/ffmpeg")]);
|
|
const cap = compiled.status === 0 ? pkg().sandboxCapability({ rootfs, cgroupRoot: process.env.MEDIA_SANDBOX_CGROUP_ROOT ?? "/sys/fs/cgroup" }) : { available: false, reason: "static compiler unavailable" };
|
|
if (!cap.available) { t.skip(`rootless sandbox canary skipped: ${cap.reason}`); return; }
|
|
}
|
|
const canaries = ["stream-key-canary-supervisor-531", "oauth-token-canary-supervisor-794", "srt-passphrase-canary-supervisor-652"];
|
|
await env({ NODE_ENV: "development", MEDIA_SANDBOX: enabled ? "on" : "off", MEDIA_SANDBOX_ROOTFS: rootfs,
|
|
MEDIA_SANDBOX_CGROUP_ROOT: process.env.MEDIA_SANDBOX_CGROUP_ROOT ?? "/sys/fs/cgroup", MEDIA_SCRATCH_DIR: join(dir, "scratch"),
|
|
CANARY_STREAM_KEY: canaries[0], CANARY_OAUTH_TOKEN: canaries[1], CANARY_SRT_PASSPHRASE: canaries[2] }, async () => {
|
|
const keys = generateKeyPairSync("ed25519"); const lines: string[] = []; let proc: ManagedProcess | undefined;
|
|
const worker = new MediaSupervisor(channel.workerId, [channel], store, new DurableFence(join(dir, "fence")), new Map([["control", keys.publicKey]]),
|
|
(_channel, onExit): RelayJob => ({
|
|
async start() {
|
|
proc = spawnManaged("ffmpeg", ["-hide_banner", "-nostdin", "-loglevel", "error", "-f", "mpegts", "-i", "pipe:0", "-f", "null", "pipe:1"],
|
|
["pipe", "ignore", "pipe"], { redact: value => value }, line => lines.push(line));
|
|
void proc.exited.then(({ code }) => onExit(`ffmpeg exited code=${code}`)); await proc.started;
|
|
},
|
|
async stop() { await proc?.terminate(100); },
|
|
}), { transition: event => lines.push(JSON.stringify(event)) }, { schedule() { return { cancel() {} }; } });
|
|
try {
|
|
await worker.start(); await worker.handle(command(keys));
|
|
assert.equal(worker.status()[0]?.state, "RUNNING"); assert.ok(proc?.child.pid);
|
|
assert.equal(proc.child.spawnfile.endsWith("media-confine"), enabled);
|
|
const cmdline = readFileSync(`/proc/${proc.child.pid}/cmdline`, "utf8");
|
|
const environment = readFileSync(`/proc/${proc.child.pid}/environ`, "utf8");
|
|
proc.child.stdin!.end("invalid-media-to-force-ffmpeg-error");
|
|
const result = await proc.exited; assert.notEqual(result.code, 0);
|
|
await worker.stop();
|
|
assert.ok(lines.some(line => enabled ? line.includes("stderr suppressed") : /Invalid data|could not find codec|End of file/i.test(line)));
|
|
for (const canary of canaries) for (const output of [cmdline, environment, ...lines]) assert.ok(!output.includes(canary));
|
|
} finally { await worker.stop(); }
|
|
});
|
|
} finally { rmSync(dir, { recursive: true, force: true }); }
|
|
});
|