josh/ytplayer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Add device diagnostics and P2P access controls

Browse Source
This commit is contained in:
Jonathan Sykes
2026-10-14 02:45:33 +08:00
parent e9a597c415
commit 5177a4d441
11 changed files with 147 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

35
server/p2p-admin.js Normal file
View File

@@ -0,0 +1,35 @@
import { audit } from './audit-log.js';
export function registerP2pAdminRoutes(app, { db, adminAuth, p2pDb, hub, stats = async () => ({}) }) {
app.get('/api/admin/p2p/devices', adminAuth, async c => {
const offset = Math.max(0, Number(c.req.query('offset')) || 0);
const devices = await p2pDb.listAdminDevices({ limit: 100, offset });
return c.json({ ok: true, devices: devices.map(d => ({
id: d.device_id, fingerprint: d.fingerprint || '', fingerprintShort: (d.fingerprint || d.device_id).slice(0, 12),
profile: d.profile || '', userAgent: d.user_agent || '', share: !!Number(d.share), banned: !!d.banned_at,
lastSeenAt: Number(d.last_seen_at), online: hub.isOnline(d.device_id), holdings: Number(d.holdings), holdingBytes: Number(d.holding_bytes),
bytesUp: Number(d.bytes_up), bytesDown: Number(d.bytes_down),
})), network: { online: hub.onlineCount(), ...(await stats()) }, offset }, 200, { 'Cache-Control': 'no-store' });
});
app.post('/api/admin/p2p/devices/:id/ban', adminAuth, async c => {
const id = c.req.param('id'); if (!/^dev_[0-9a-f]{16}$/.test(id)) return c.json({ ok: false, error: 'Invalid device.' }, 400);
const body = await c.req.json().catch(() => ({})); const banned = body.banned === true;
const changed = await p2pDb.setDeviceBanned(id, banned);
if (!changed) return c.json({ ok: false, error: 'Device not found.' }, 404);
if (banned) hub.disconnect(id);
await audit(db, 'admin', banned ? 'p2p.device.ban' : 'p2p.device.unban', id);
return c.json({ ok: true, banned });
});
app.post('/api/admin/p2p/devices/:id/reannounce', adminAuth, async c => {
const id = c.req.param('id'); if (!/^dev_[0-9a-f]{16}$/.test(id)) return c.json({ ok: false, error: 'Invalid device.' }, 400);
const sent = hub.send(id, { type: 'force-sync' });
await audit(db, 'admin', 'p2p.device.reannounce', id, { sent });
return c.json({ ok: true, sent });
});
app.delete('/api/admin/p2p/devices/:id/holdings', adminAuth, async c => {
const id = c.req.param('id'); if (!/^dev_[0-9a-f]{16}$/.test(id)) return c.json({ ok: false, error: 'Invalid device.' }, 400);
const removed = await p2pDb.removeDeviceHolders(id);
await audit(db, 'admin', 'p2p.device.holdings.revoke', id, { removed });
return c.json({ ok: true, removed });
});
}

36
server/p2p-db.js
View File

@@ -37,6 +37,10 @@ export async function initP2pSchema() {
fingerprint TEXT,
profile TEXT,
share INTEGER NOT NULL DEFAULT 1,
user_agent TEXT,
banned_at INTEGER,
bytes_up INTEGER NOT NULL DEFAULT 0,
bytes_down INTEGER NOT NULL DEFAULT 0,
created_at INTEGER NOT NULL,
last_seen_at INTEGER NOT NULL
);
@@ -60,6 +64,10 @@ export async function initP2pSchema() {
PRIMARY KEY (video_id, day)
);
`);
for (const [name, decl] of [['user_agent','TEXT'],['banned_at','INTEGER'],['bytes_up','INTEGER NOT NULL DEFAULT 0'],['bytes_down','INTEGER NOT NULL DEFAULT 0']]) {
try { await db.execute(`ALTER TABLE p2p_devices ADD COLUMN ${name} ${decl}`); }
catch (e) { if (!/duplicate column/i.test(String(e.message))) throw e; }
}
// media_cache.sha256 — the cid of the current <id>.<gen>.mp4 (plan 009).
try { await db.execute('ALTER TABLE media_cache ADD COLUMN sha256 TEXT'); }
catch (e) { if (!/duplicate column/i.test(String(e.message))) throw e; }
@@ -117,11 +125,11 @@ export async function revokeContent(cid) {
// ---- devices ------------------------------------------------------------------
export async function createDevice({ deviceId, secretHash, fingerprint, profile, now }) {
export async function createDevice({ deviceId, secretHash, fingerprint, profile, userAgent, now }) {
await db.execute({
sql: `INSERT INTO p2p_devices (device_id, secret_hash, fingerprint, profile, share, created_at, last_seen_at)
VALUES (?, ?, ?, ?, 1, ?, ?)`,
args: [deviceId, secretHash, fingerprint || null, profile || null, now, now],
sql: `INSERT INTO p2p_devices (device_id, secret_hash, fingerprint, profile, share, user_agent, created_at, last_seen_at)
VALUES (?, ?, ?, ?, 1, ?, ?, ?)`,
args: [deviceId, secretHash, fingerprint || null, profile || null, userAgent || null, now, now],
});
}
@@ -139,6 +147,26 @@ export async function touchDevice(deviceId, { now, share, profile } = {}) {
});
}
export async function listAdminDevices({ limit = 100, offset = 0 } = {}) {
const r = await db.execute({ sql: `SELECT d.device_id,d.fingerprint,d.profile,d.share,d.user_agent,d.banned_at,d.bytes_up,d.bytes_down,d.created_at,d.last_seen_at,
COUNT(CASE WHEN h.status='active' THEN 1 END) AS holdings,
COALESCE(SUM(CASE WHEN h.status='active' THEN c.size ELSE 0 END),0) AS holding_bytes
FROM p2p_devices d LEFT JOIN p2p_holders h ON h.device_id=d.device_id LEFT JOIN p2p_content c ON c.cid=h.cid
GROUP BY d.device_id ORDER BY d.last_seen_at DESC LIMIT ? OFFSET ?`, args: [Math.max(1, Math.min(200, limit)), Math.max(0, offset)] });
return rowsOf(r);
}
export async function setDeviceBanned(deviceId, banned, now = Date.now()) {
const r = await db.execute({ sql: 'UPDATE p2p_devices SET banned_at=? WHERE device_id=?', args: [banned ? now : null, deviceId] });
return !!r.rowsAffected;
}
export async function removeDeviceHolders(deviceId, now = Date.now()) {
const r = await db.execute({ sql: "UPDATE p2p_holders SET status='removed',removed_at=? WHERE device_id=? AND status='active'", args: [now, deviceId] });
return Number(r.rowsAffected || 0);
}
export async function recordDeviceTraffic(deviceId, { up = 0, down = 0 } = {}) {
await db.execute({ sql: 'UPDATE p2p_devices SET bytes_up=bytes_up+?,bytes_down=bytes_down+? WHERE device_id=?', args: [Math.max(0, Number(up) || 0), Math.max(0, Number(down) || 0), deviceId] });
}
// ---- holders (persistent; never expired by time) ------------------------------
export async function upsertHolder({ cid, deviceId, trust = 'reported', now }) {

20
server/p2p-db.test.js
View File

@@ -40,6 +40,22 @@ test('content upsert/get/known/revoke', async () => {
expect([...(await P.knownCids([CID2]))]).toEqual([]);
});
test('admin device inventory, ban state, and holding revocation', async () => {
const id = 'dev_00000000000000b1';
await P.createDevice({ deviceId: id, secretHash: 'hash', fingerprint: 'phone-fp', profile: 'family', userAgent: 'iPhone', now: T0 });
const adminCid = 'c'.repeat(64);
await P.upsertContent({ cid: adminCid, videoId: 'dQw4w9WgXcQ', size: 4096, origin: 'server', now: T0 });
await P.upsertHolder({ cid: adminCid, deviceId: id, now: T0 });
let [device] = await P.listAdminDevices();
expect(device).toMatchObject({ device_id: id, user_agent: 'iPhone', holdings: 1, holding_bytes: 4096 });
expect(await P.setDeviceBanned(id, true, T0 + 100)).toBe(true);
[device] = await P.listAdminDevices(); expect(device.banned_at).toBe(T0 + 100);
expect(await P.removeDeviceHolders(id, T0 + 200)).toBe(1);
[device] = await P.listAdminDevices(); expect(device.holdings).toBe(0);
expect(await P.setDeviceBanned(id, false, T0 + 300)).toBe(true);
[device] = await P.listAdminDevices(); expect(device.banned_at).toBeNull();
});
test('holders persist, never expire, and a full report removes missing ones', async () => {
await P.createDevice({ deviceId: 'dev_1', secretHash: 'h', fingerprint: 'fp', now: T0 });
await P.upsertHolder({ cid: CID, deviceId: 'dev_1', now: T0 });
@@ -66,8 +82,8 @@ test('views per day and window sums', async () => {
expect(await P.viewsSince('vid00000001', T0 - 30 * 86400_000)).toBe(2);
expect(await P.viewsSince('vid00000001', T0 - 50 * 86400_000)).toBe(3);
const s = await P.p2pStats();
expect(s.content).toBe(1);
expect(s.devices).toBe(1);
expect(s.content).toBe(2);
expect(s.devices).toBe(2);
});
test('listSharedContent: newest verified file per video, only with sharing holders counted', async () => {

4
server/p2p-hub.js
View File

@@ -47,7 +47,7 @@ export function createP2pHub({ getDevice, authorizeProfile = async () => '', ena
const deviceId = String(m.device || '');
const secret = String(m.secret || '');
const d = /^dev_[0-9a-f]{16}$/.test(deviceId) ? await getDevice(deviceId).catch(() => null) : null;
if (!d || !same(d.secret_hash, sha(secret))) { try { ws.close(4401, 'unknown device'); } catch { /* gone */ } return; }
if (!d || d.banned_at || !same(d.secret_hash, sha(secret))) { try { ws.close(d?.banned_at ? 4403 : 4401, d?.banned_at ? 'device banned' : 'unknown device'); } catch { /* gone */ } return; }
ws.data.profile = await authorizeProfile(m.profile, m.profileSecret, deviceId).catch(() => '');
ws.data.inventory = [];
ws.data.authed = true;
@@ -107,6 +107,8 @@ export function createP2pHub({ getDevice, authorizeProfile = async () => '', ena
websocket: { open, message, close },
isOnline: (deviceId) => online.has(deviceId),
send: (deviceId, msg) => { const ws = online.get(deviceId); if (!ws) return false; send(ws, msg); return true; },
disconnect: (deviceId) => { const ws = online.get(deviceId); if (!ws) return false; try { ws.close(4403, 'device banned'); } catch { /* gone */ } return true; },
onlineDevices: () => [...online.keys()],
onlineCount: () => online.size,
};
}

15
server/p2p-hub.test.js
View File

@@ -51,6 +51,21 @@ test('auth on open, hello with opaque peer id, bad secret is closed', async () =
expect(hub.isOnline('dev_000000000000000b')).toBe(false);
});
test('banned devices cannot authenticate and admin disconnect closes an online peer', async () => {
await p2pDb.setDeviceBanned('dev_000000000000000a', true, NOW);
const hub = createP2pHub({ getDevice: p2pDb.getDevice });
const banned = fakeWs({ budget: [] });
await hub.websocket.message(banned, JSON.stringify({ type: 'auth', device: 'dev_000000000000000a', secret: SECRET_A }));
expect(banned.closed).toBe(4403);
await p2pDb.setDeviceBanned('dev_000000000000000a', false, NOW + 1);
const online = fakeWs({ budget: [] });
await hub.websocket.message(online, JSON.stringify({ type: 'auth', device: 'dev_000000000000000a', secret: SECRET_A }));
expect(hub.disconnect('dev_000000000000000a')).toBe(true);
expect(online.closed).toBe(4403);
hub.websocket.close(online);
expect(hub.isOnline('dev_000000000000000a')).toBe(false);
});
test('legacy unconfirmed signalling is refused; close drops presence', async () => {
const hub = createP2pHub({ getDevice: p2pDb.getDevice });
const a = fakeWs({ budget: [] });

1
server/p2p-intake.js
View File

@@ -105,6 +105,7 @@ export function registerIntakeRoutes(app, deps) {
);
if (!adm.ok) throw Object.assign(new Error('not admitted: ' + adm.reason), { status: 422 });
await p2pDb.upsertHolder({ cid, deviceId: d.device_id, trust: 'challenged', now: now() });
await p2pDb.recordDeviceTraffic?.(d.device_id, { up: got });
let adopted = false;
if (adopt) {
try { adopted = !!(await adopt(tk.videoId, path, { sha256: cid, probe, meta: tk.meta })).adopted; keep = adopted; }

3
server/p2p-routes.js
View File

@@ -41,10 +41,12 @@ export function registerP2pRoutes(app, deps) {
if (!m) return null;
const d = await p2pDb.getDevice(m[1]);
if (!d || !same(d.secret_hash, sha(m[2]))) return null;
if (d.banned_at) return { ...d, _banned: true };
return d;
}
const requireDevice = async (c, next) => {
const d = await deviceOf(c);
if (d?._banned) return c.json({ ok: false, error: 'device access revoked' }, 403);
if (!d) return c.json({ ok: false, error: 'unknown device' }, 401);
c.set('device', d);
await next();
@@ -67,6 +69,7 @@ export function registerP2pRoutes(app, deps) {
deviceId, secretHash: sha(secret),
fingerprint: String(body.fingerprint || '').slice(0, 128) || null,
profile: String(body.profile || '').slice(0, 64) || null,
userAgent: String(c.req.header('user-agent') || '').slice(0, 500),
now: t,
});
return c.json({ ok: true, deviceId, secret });

9
server/p2p-routes.test.js
View File

@@ -63,6 +63,15 @@ test('holdings need a valid device secret', async () => {
expect((await req('/api/p2p/holdings', { method: 'POST', body: { items: [] }, dev: { ...dev, secret: 'e'.repeat(64) } })).status).toBe(401);
});
test('banned device is denied with an explicit forbidden response', async () => {
const dev = await newDevice();
await p2pDb.setDeviceBanned(dev.deviceId, true);
const r = await req('/api/p2p/holdings', { method: 'POST', body: { items: [] }, dev });
expect(r.status).toBe(403);
expect(await r.json()).toMatchObject({ ok: false, error: 'device access revoked' });
await p2pDb.setDeviceBanned(dev.deviceId, false);
});
test('report accepts verified cids, returns unknown ones, and challenges the server-held file', async () => {
const dev = await newDevice();
const r = await (await req('/api/p2p/holdings', { method: 'POST', dev, body: { share: true, items: [

3
server/server.js
View File

@@ -72,6 +72,7 @@ import { createWarmQueue } from './warm-queue.js';
import { registerDeepSearchRoutes } from './deep-search.js';
import { dirname, join as pathJoin } from 'node:path';
import { createRequestMonitor, registerRequestMonitorRoutes } from './request-monitor.js';
import { registerP2pAdminRoutes } from './p2p-admin.js';
// A media proxy must not die because one client's stream hit an edge case
// (see /api/play cancel()): log and keep serving instead of crash-looping.
@@ -2381,6 +2382,8 @@ const p2pHub = createP2pHub({ getDevice: p2pDb.getDevice, enabled: () => P2P.ena
directProfileFails.set(key, { count: old && Date.now() - old.at < 900000 ? old.count + 1 : 1, at: Date.now() });
if (directProfileFails.size > 5000) directProfileFails.clear(); return '';
} });
registerP2pAdminRoutes(app, { db: notesDb, adminAuth: notes.requireAdminOrToken, p2pDb, hub: p2pHub,
stats: async () => ({ intakeOpen: intake.openTickets(), intakeActive: intake.active(), content: (await p2pDb.p2pStats()).content }) });
// GET /api/p2p/holders?v=<videoId>|cid=<sha256> — who holds a copy. Rows are
// persistent; `stale` flags a holder not re-verified for P2P_STALE_DAYS.